Digital Drosophila Archive
495 lines
31 KiB
Python
495 lines
31 KiB
Python
"""Pure Phase 4AN proposal metrics; no file I/O, simulation or fitted decoder.
|
|
|
|
The caller must bind complete anatomical population membership (not a plastic
|
|
mask subset), finalized execution receipts, input identities and seed lineages
|
|
to its preregistration. These functions validate arrays/declared coverage; they
|
|
cannot establish provenance by trusting a caller's Boolean or SHA string.
|
|
All gates below are NEW DRAFT hypotheses, not a historical learning result.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from collections.abc import Mapping
|
|
|
|
import numpy as np
|
|
|
|
|
|
SCHEMA = 'phase4an-metrics-draft-v1'
|
|
CUES = ('4-ethylphenol', '3-hexanone')
|
|
MODES = ('physiological', 'matched_energy')
|
|
DAN_TYPES = ('PAM15', 'PPL103')
|
|
CONTROLS = ('D=0', 'frozen', 'random reward', 'unpaired', 'CS-only', 'DAN-only')
|
|
ARMS = ('paired',) + CONTROLS
|
|
POPULATIONS = ('MBON', 'legacyDownstream', 'nonMBONDownstream', 'MBON01', 'MBON12')
|
|
SEPARATION_POPULATIONS = ('MBON', 'legacyDownstream', 'nonMBONDownstream')
|
|
POPULATION_SEMANTICS = {
|
|
'MBON': 'complete caller-bound anatomical MBON population',
|
|
'legacyDownstream': 'complete frozen H downstream membership, including MBONs; never mask-filtered',
|
|
'nonMBONDownstream': 'exact legacyDownstream minus MBON; NEW hypothesis readout, not the old H population',
|
|
'MBON01': 'complete caller-bound MBON01 type, not plastic-mask targets only',
|
|
'MBON12': 'complete caller-bound MBON12 type, not plastic-mask targets only',
|
|
}
|
|
STATE_KEYS = frozenset(('step', 'neuron_idx', 'counts', 'v', 'g', 'ref'))
|
|
|
|
|
|
def require(condition, message):
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def _hash(value):
|
|
return hashlib.sha256(np.ascontiguousarray(value, dtype=value.dtype.newbyteorder('<')).tobytes()).hexdigest()
|
|
|
|
|
|
def _json_hash(value):
|
|
raw = json.dumps(value, sort_keys=True, separators=(',', ':'), allow_nan=False).encode('utf-8')
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def _sha(value):
|
|
return type(value) is str and len(value) == 64 and all(c in '0123456789abcdef' for c in value)
|
|
|
|
|
|
def _token(value, name):
|
|
require(type(value) is str and bool(value) and value.strip() == value, 'Invalid ' + name)
|
|
return value
|
|
|
|
|
|
def _vector(value, dtype, name, size=None):
|
|
require(isinstance(value, np.ndarray) and value.dtype == np.dtype(dtype) and value.ndim == 1,
|
|
name + ' requires an exact ' + dtype + ' vector')
|
|
require(size is None or value.shape == (size,), name + ' shape mismatch')
|
|
return value
|
|
|
|
|
|
def _ids(value, name):
|
|
_vector(value, 'int32', name)
|
|
require(len(value) > 0 and np.all((value >= 0) & (value < 166700)) and np.all(value[1:] > value[:-1]),
|
|
name + ' must be nonempty sorted unique global neuron indices')
|
|
return value
|
|
|
|
|
|
def _positions(nodes, values, name):
|
|
position = np.searchsorted(nodes, values)
|
|
require(np.all(position < len(nodes)), name + ' contains absent neurons')
|
|
require(np.array_equal(nodes[position], values), name + ' contains absent neurons')
|
|
return position
|
|
|
|
|
|
def validate_populations(nodes, populations):
|
|
"""Validate full caller-bound type lists; never derive them from an edge mask."""
|
|
_ids(nodes, 'nodes')
|
|
require(isinstance(populations, Mapping) and set(populations) == set(POPULATIONS),
|
|
'Require exactly MBON/legacyDownstream/nonMBONDownstream/MBON01/MBON12 memberships')
|
|
for name in POPULATIONS:
|
|
_positions(nodes, _ids(populations[name], name), name)
|
|
for name in ('MBON01', 'MBON12'):
|
|
_positions(populations['MBON'], populations[name], name + ' outside MBON')
|
|
require(not len(np.intersect1d(populations['MBON01'], populations['MBON12'])), 'Type populations overlap')
|
|
_positions(populations['legacyDownstream'], populations['MBON'], 'MBON outside legacyDownstream')
|
|
require(np.array_equal(populations['nonMBONDownstream'],
|
|
np.setdiff1d(populations['legacyDownstream'], populations['MBON'])),
|
|
'nonMBONDownstream must equal the complete legacyDownstream minus MBON')
|
|
payload = {name: populations[name].tolist() for name in POPULATIONS}
|
|
return dict(indices=payload, membership_sha256=_json_hash(payload),
|
|
population_semantics=dict(POPULATION_SEMANTICS),
|
|
complete_anatomical_membership_authority='caller preregistration; never inferred from mask')
|
|
|
|
|
|
def validate_state(state, *, nodes, steps, dt_ms=.1):
|
|
"""Check every original state element and all sparse events without mutation."""
|
|
_ids(nodes, 'nodes')
|
|
require(type(steps) is int and 0 < steps <= np.iinfo(np.int32).max, 'Invalid steps')
|
|
require(type(dt_ms) in (int, float) and math.isfinite(dt_ms) and dt_ms > 0, 'Invalid dt_ms')
|
|
require(isinstance(state, Mapping) and set(state) == STATE_KEYS, 'State must contain exactly six runtime keys')
|
|
n = len(nodes)
|
|
for key, dtype in (('counts', 'int32'), ('ref', 'int32'), ('v', 'float64'), ('g', 'float64')):
|
|
_vector(state[key], dtype, key, n)
|
|
require(np.isfinite(state[key]).all(), 'Nonfinite state: ' + key)
|
|
require(np.all((state['counts'] >= 0) & (state['counts'] <= steps)), 'Invalid counts')
|
|
require(np.all((state['ref'] >= 0) & (state['ref'] <= 22)), 'Invalid refractory state')
|
|
ticks = _vector(state['step'], 'int32', 'step')
|
|
events = _vector(state['neuron_idx'], 'int32', 'neuron_idx', len(ticks))
|
|
require(np.all((ticks >= 0) & (ticks < steps)), 'Event tick outside trial')
|
|
require(len(ticks) < 2 or np.all((ticks[1:] > ticks[:-1]) |
|
|
((ticks[1:] == ticks[:-1]) & (events[1:] > events[:-1]))),
|
|
'Events must be sorted unique by tick/global neuron')
|
|
local = _positions(nodes, events, 'events')
|
|
require(np.array_equal(np.bincount(local, minlength=n), state['counts']), 'Sparse events/counts mismatch')
|
|
first = np.full(n, -1, dtype=np.int32)
|
|
if len(ticks):
|
|
first.fill(steps)
|
|
np.minimum.at(first, local, ticks)
|
|
first[first == steps] = -1
|
|
return dict(first_tick=first, local_events=local,
|
|
hashes={key: _hash(state[key]) for key in sorted(STATE_KEYS)})
|
|
|
|
|
|
def _first_ms(first, dt_ms):
|
|
present = first[first >= 0]
|
|
return float(int(present.min()) * dt_ms) if len(present) else None
|
|
|
|
|
|
def _latencies(first, dt_ms):
|
|
return [None if int(tick) < 0 else float(int(tick) * dt_ms) for tick in first]
|
|
|
|
|
|
def _population_change(pre, post, checked_pre, checked_post, nodes, local, dt_ms, vectors):
|
|
cp, cq = pre['counts'][local].astype(np.int64), post['counts'][local].astype(np.int64)
|
|
fp, fq = checked_pre['first_tick'][local], checked_post['first_tick'][local]
|
|
lp, lq = _first_ms(fp, dt_ms), _first_ms(fq, dt_ms)
|
|
both = (fp >= 0) & (fq >= 0)
|
|
ep = np.isin(checked_pre['local_events'], local)
|
|
eq = np.isin(checked_post['local_events'], local)
|
|
events_same = (np.array_equal(pre['step'][ep], post['step'][eq]) and
|
|
np.array_equal(pre['neuron_idx'][ep], post['neuron_idx'][eq]))
|
|
try:
|
|
with np.errstate(over='raise', invalid='raise'):
|
|
dv, dg = post['v'][local] - pre['v'][local], post['g'][local] - pre['g'][local]
|
|
vmax, gmax = float(np.max(np.abs(dv))), float(np.max(np.abs(dg)))
|
|
except FloatingPointError as error:
|
|
raise ValueError('Nonfinite derived final-state difference') from error
|
|
result = dict(neurons=len(local), pre_total_spikes=int(cp.sum()), post_total_spikes=int(cq.sum()),
|
|
spike_count_l1=int(np.abs(cq - cp).sum()), spike_count_changed_neurons=int(np.count_nonzero(cp != cq)),
|
|
events_changed=not events_same, pre_first_latency_ms=lp, post_first_latency_ms=lq,
|
|
first_latency_delta_ms=None if lp is None or lq is None else float(lq - lp),
|
|
defined_first_latency_changed=bool(lp is not None and lq is not None and lp != lq),
|
|
first_latency_both_defined_neurons=int(both.sum()),
|
|
first_latency_changed_among_both_defined_neurons=int(np.count_nonzero(both & (fp != fq))),
|
|
first_latency_became_defined_neurons=int(np.count_nonzero((fp < 0) & (fq >= 0))),
|
|
first_latency_became_undefined_neurons=int(np.count_nonzero((fp >= 0) & (fq < 0))),
|
|
final_delta_v_max_abs=vmax, final_delta_g_max_abs=gmax,
|
|
final_ref_changed_neurons=int(np.count_nonzero(pre['ref'][local] != post['ref'][local])),
|
|
final_state_units='SI volts; refractory integer ticks',
|
|
max_abs_scope='maximum across population neurons at terminal tick; NOT a temporal peak')
|
|
if vectors:
|
|
result.update(neuron_idx=nodes[local].tolist(), pre_counts=cp.tolist(), post_counts=cq.tolist(),
|
|
pre_first_latency_ms_per_neuron=_latencies(fp, dt_ms),
|
|
post_first_latency_ms_per_neuron=_latencies(fq, dt_ms))
|
|
for key in ('v', 'g', 'ref'):
|
|
result['pre_final_' + key] = pre[key][local].tolist()
|
|
result['post_final_' + key] = post[key][local].tolist()
|
|
return result
|
|
|
|
|
|
def functional_summary(pre, post, *, nodes, populations, steps, dt_ms=.1):
|
|
"""Full-state validation plus wholebrain and anatomical-group differences.
|
|
|
|
The population's earliest event defines the legacy-style first latency.
|
|
Per-neuron first events and missingness transitions are additional diagnostics.
|
|
A missing latency is None, never zero or evidence of a defined latency shift.
|
|
"""
|
|
membership = validate_populations(nodes, populations)
|
|
a = validate_state(pre, nodes=nodes, steps=steps, dt_ms=dt_ms)
|
|
b = validate_state(post, nodes=nodes, steps=steps, dt_ms=dt_ms)
|
|
groups = {name: _population_change(pre, post, a, b, nodes,
|
|
_positions(nodes, populations[name], name), dt_ms, True) for name in POPULATIONS}
|
|
return dict(schema=SCHEMA, full_six_key_state_validated=True, steps=steps, dt_ms=float(dt_ms),
|
|
nodes=len(nodes), nodes_raw_sha256=_hash(nodes), membership=membership,
|
|
pre_state_raw_sha256=a['hashes'], post_state_raw_sha256=b['hashes'],
|
|
wholebrain=_population_change(pre, post, a, b, nodes, np.arange(len(nodes)), dt_ms, False),
|
|
populations=groups, MBON_final_delta_g_max_abs=groups['MBON']['final_delta_g_max_abs'],
|
|
MBON_final_delta_g_units='SI volts', MBON_final_delta_g_scope='terminal across-neuron maximum, not time peak')
|
|
|
|
|
|
def arm_metrics(pre_by_cue, post_by_cue, *, conditioned_cue, nodes, populations, steps=1200, dt_ms=.1):
|
|
"""New fixed, unscaled type-count contrast; neither decoder nor feedback."""
|
|
require(conditioned_cue in CUES, 'Unknown conditioned odor label')
|
|
require(isinstance(pre_by_cue, Mapping) and isinstance(post_by_cue, Mapping) and
|
|
set(pre_by_cue) == set(post_by_cue) == set(CUES), 'Both original odor labels are required')
|
|
neutral = next(cue for cue in CUES if cue != conditioned_cue)
|
|
summaries = {cue: functional_summary(pre_by_cue[cue], post_by_cue[cue], nodes=nodes,
|
|
populations=populations, steps=steps, dt_ms=dt_ms) for cue in CUES}
|
|
b, means = {}, {}
|
|
for phase in ('pre', 'post'):
|
|
b[phase], means[phase] = {}, {}
|
|
for cue in CUES:
|
|
groups = summaries[cue]['populations']
|
|
means[phase][cue] = {name: float(np.mean(groups[name][phase + '_counts'])) for name in ('MBON01', 'MBON12')}
|
|
b[phase][cue] = means[phase][cue]['MBON12'] - means[phase][cue]['MBON01']
|
|
raw_s = (b['post'][conditioned_cue] - b['post'][neutral]) - (b['pre'][conditioned_cue] - b['pre'][neutral])
|
|
separation = {}
|
|
for name in SEPARATION_POPULATIONS:
|
|
values = {}
|
|
for phase in ('pre', 'post'):
|
|
first = np.asarray(summaries[CUES[0]]['populations'][name][phase + '_counts'], dtype=np.float64)
|
|
second = np.asarray(summaries[CUES[1]]['populations'][name][phase + '_counts'], dtype=np.float64)
|
|
values[phase] = float(np.linalg.norm(first - second))
|
|
separation[name] = dict(values, gain=values['post'] - values['pre'],
|
|
vector='all bound member spike counts', population_semantics=POPULATION_SEMANTICS[name])
|
|
return dict(schema=SCHEMA, analysis_status='DRAFT_NEW_HYPOTHESIS', conditioned_cue=conditioned_cue,
|
|
neutral_cue=neutral, steps=steps, dt_ms=float(dt_ms), rawB=b, rawS=float(raw_s),
|
|
type_mean_counts=means, B_definition='mean full MBON12 counts - mean full MBON01 counts',
|
|
B_is_decoder=False, S_orientation='conditioned minus neutral; post minus own pre',
|
|
separation=separation, cues=summaries,
|
|
historical_implementation_equivalence_claimed=False, learning_pass_claimed=False)
|
|
|
|
|
|
def _finite_tree(value):
|
|
if isinstance(value, Mapping):
|
|
for item in value.values():
|
|
_finite_tree(item)
|
|
elif isinstance(value, (list, tuple)):
|
|
for item in value:
|
|
_finite_tree(item)
|
|
elif isinstance(value, (float, np.floating)):
|
|
require(math.isfinite(float(value)), 'Nonfinite metric value')
|
|
|
|
|
|
def _bound_metric(metric, cue):
|
|
require(isinstance(metric, Mapping) and metric.get('schema') == SCHEMA and
|
|
metric.get('analysis_status') == 'DRAFT_NEW_HYPOTHESIS', 'Unexpected metric schema')
|
|
_finite_tree(metric)
|
|
require(metric['conditioned_cue'] == cue and metric['neutral_cue'] == next(c for c in CUES if c != cue),
|
|
'Cue role mismatch')
|
|
require(metric['steps'] == 1200 and metric['dt_ms'] == .1, 'Evaluation must use the fixed 120 ms probe')
|
|
# Derive gate quantities from recorded raw vectors, not a supplied gate Boolean.
|
|
for odor in CUES:
|
|
summary = metric['cues'][odor]
|
|
require(summary.get('full_six_key_state_validated') is True and summary['steps'] == 1200 and
|
|
summary['dt_ms'] == .1 and _sha(summary['nodes_raw_sha256']), 'Unvalidated cue state')
|
|
membership = summary['membership']['indices']
|
|
require(set(membership) == set(POPULATIONS) and summary['membership']['membership_sha256'] == _json_hash(membership),
|
|
'Population membership digest mismatch')
|
|
require(summary['membership'].get('population_semantics') == POPULATION_SEMANTICS,
|
|
'Population semantics must distinguish old H membership from the new non-MBON readout')
|
|
for key in ('pre_state_raw_sha256', 'post_state_raw_sha256'):
|
|
require(set(summary[key]) == STATE_KEYS and all(_sha(v) for v in summary[key].values()), 'Missing state digest')
|
|
for name in POPULATIONS:
|
|
group = summary['populations'][name]
|
|
ids, count = group['neuron_idx'], group['neurons']
|
|
require(type(count) is int and count > 0 and len(ids) == count and ids == membership[name] and
|
|
all(type(v) is int and 0 <= v < 166700 for v in ids) and ids == sorted(set(ids)),
|
|
'Empty or mismatched anatomical population')
|
|
for phase in ('pre', 'post'):
|
|
counts = group[phase + '_counts']
|
|
latency = group[phase + '_first_latency_ms_per_neuron']
|
|
require(len(counts) == len(latency) == count and
|
|
all(type(v) is int and 0 <= v <= 1200 for v in counts), 'Invalid count/latency vector')
|
|
for c, value in zip(counts, latency):
|
|
require((c == 0 and value is None) or (c > 0 and type(value) in (int, float) and
|
|
0 <= value < 120 and value == round(value / .1) * .1), 'Missing/nonphysical first latency')
|
|
present = [v for v in latency if v is not None]
|
|
require(group[phase + '_first_latency_ms'] == (min(present) if present else None),
|
|
'Population first latency differs from raw cells')
|
|
for key in ('v', 'g', 'ref'):
|
|
values = group[phase + '_final_' + key]
|
|
require(len(values) == count and all(type(v) in (int, float) and math.isfinite(v) for v in values),
|
|
'Invalid final-state vector')
|
|
if key == 'ref':
|
|
require(all(type(v) is int and 0 <= v <= 22 for v in values), 'Invalid final ref')
|
|
l1 = sum(abs(a - b) for a, b in zip(group['pre_counts'], group['post_counts']))
|
|
lp, lq = group['pre_first_latency_ms'], group['post_first_latency_ms']
|
|
changed = lp is not None and lq is not None and lp != lq
|
|
require(type(group['spike_count_l1']) is int and group['spike_count_l1'] == l1 and
|
|
group['defined_first_latency_changed'] is changed and
|
|
group['first_latency_delta_ms'] == (None if lp is None or lq is None else lq - lp),
|
|
'Functional predicate differs from raw counts/defined latency')
|
|
try:
|
|
with np.errstate(over='raise', invalid='raise'):
|
|
for key in ('v', 'g'):
|
|
difference = np.asarray(group['post_final_' + key]) - np.asarray(group['pre_final_' + key])
|
|
require(group['final_delta_' + key + '_max_abs'] == float(np.max(np.abs(difference))),
|
|
'Final-state maximum differs from raw terminal vectors')
|
|
except FloatingPointError as error:
|
|
raise ValueError('Nonfinite derived terminal metric') from error
|
|
require(not set(membership['MBON01']).intersection(membership['MBON12']), 'Type population overlap')
|
|
require(set(membership['MBON']) <= set(membership['legacyDownstream']),
|
|
'Legacy downstream must retain the complete MBON population')
|
|
require(membership['nonMBONDownstream'] == sorted(set(membership['legacyDownstream']) - set(membership['MBON'])),
|
|
'nonMBONDownstream differs from exact legacyDownstream minus MBON')
|
|
for name, parent in (('MBON01', 'MBON'), ('MBON12', 'MBON'),
|
|
('MBON', 'legacyDownstream'), ('nonMBONDownstream', 'legacyDownstream')):
|
|
require(set(membership[name]) <= set(membership[parent]), 'Population is not a subset of its bound parent')
|
|
lookup = {v: i for i, v in enumerate(membership[parent])}
|
|
for phase in ('pre', 'post'):
|
|
for field in ('counts', 'first_latency_ms_per_neuron', 'final_v', 'final_g', 'final_ref'):
|
|
key = phase + '_' + field
|
|
require(summary['populations'][name][key] == [
|
|
summary['populations'][parent][key][lookup[v]] for v in membership[name]],
|
|
'Overlapping population vectors disagree with the bound parent')
|
|
for phase in ('pre', 'post'):
|
|
for odor in CUES:
|
|
groups = metric['cues'][odor]['populations']
|
|
means = []
|
|
for name in ('MBON12', 'MBON01'):
|
|
counts = groups[name][phase + '_counts']
|
|
require(len(counts) > 0 and all(type(v) is int and 0 <= v <= 1200 for v in counts),
|
|
'Empty or invalid type count vector')
|
|
means.append(sum(counts) / len(counts))
|
|
require(metric['type_mean_counts'][phase][odor][name] == means[-1], 'Type mean differs from raw counts')
|
|
require(metric['rawB'][phase][odor] == means[0] - means[1], 'Raw B disagrees with full type vectors')
|
|
neutral = metric['neutral_cue']
|
|
calculated = (metric['rawB']['post'][cue] - metric['rawB']['post'][neutral]) - (
|
|
metric['rawB']['pre'][cue] - metric['rawB']['pre'][neutral])
|
|
require(type(metric['rawS']) in (int, float) and metric['rawS'] == calculated, 'Raw S disagrees with signed components')
|
|
for name in SEPARATION_POPULATIONS:
|
|
calculated = {}
|
|
for phase in ('pre', 'post'):
|
|
a = metric['cues'][CUES[0]]['populations'][name][phase + '_counts']
|
|
b = metric['cues'][CUES[1]]['populations'][name][phase + '_counts']
|
|
require(metric['cues'][CUES[0]]['membership'] == metric['cues'][CUES[1]]['membership'],
|
|
'A/B population mismatch')
|
|
calculated[phase] = float(np.linalg.norm(np.asarray(a, dtype=np.float64) - np.asarray(b, dtype=np.float64)))
|
|
require(metric['separation'][name]['pre'] == calculated['pre'] and
|
|
metric['separation'][name]['post'] == calculated['post'] and
|
|
metric['separation'][name]['gain'] == calculated['post'] - calculated['pre'],
|
|
'Separation gain differs from raw count vectors')
|
|
require(metric['separation'][name].get('population_semantics') == POPULATION_SEMANTICS[name],
|
|
'Separation population semantics mismatch')
|
|
return metric
|
|
|
|
|
|
def _execution(arm, name, assay, used_receipts):
|
|
require(isinstance(arm, Mapping) and set(arm) == {'metrics', 'execution'}, 'Arm needs metrics and actual execution evidence')
|
|
e = arm['execution']
|
|
required = {'complete', 'recording_verified', 'receipt_sha256', 'assay', 'epoch_duration_ms',
|
|
'cs_duration_ms', 'us_tick', 'completed_epochs', 'reinforcement_opportunities', 'terminal_pairings'}
|
|
require(isinstance(e, Mapping) and set(e) == required, 'Incomplete execution evidence')
|
|
require(e['complete'] is True and e['recording_verified'] is True and _sha(e['receipt_sha256']),
|
|
'All arms require an actual verified completed execution receipt')
|
|
require(e['receipt_sha256'] not in used_receipts, 'Execution receipt reused across arms/replicates')
|
|
used_receipts.add(e['receipt_sha256'])
|
|
require(e['assay'] == assay and e['epoch_duration_ms'] == (120. if assay == 'normal' else 180.) and
|
|
e['cs_duration_ms'] == 100. and type(e['us_tick']) is int and
|
|
e['us_tick'] == (900 if assay == 'normal' else 1500), 'Normal/delayed control timing mismatch')
|
|
require(type(e['completed_epochs']) is int and e['completed_epochs'] == (150 if name == 'unpaired' else 100),
|
|
'Incomplete arm epoch inventory')
|
|
require(type(e['reinforcement_opportunities']) is int and e['reinforcement_opportunities'] == (0 if name == 'CS-only' else 50),
|
|
'Wrong arm US opportunity inventory')
|
|
require(type(e['terminal_pairings']) is int and e['terminal_pairings'] == 50, 'Only fixed terminal count 50 may gate')
|
|
|
|
|
|
def evaluate_acquisition(replicates, *, assay, mode, dan_type, conditioned_cue, registered_training_streams):
|
|
"""Strict new normal/delayed 3-of-5 SAME-training-replicate conjunction.
|
|
|
|
Each row has replicate_id, replicate_unit='independent_training',
|
|
training_stream_ids (all independent source IDs, not aliases),
|
|
evaluation_stream_ids, assay/mode/dan_type/conditioned_cue, and seven arms.
|
|
Arm = {metrics: arm_metrics(...), execution: complete receipt declaration}.
|
|
registered_training_streams maps exactly five predeclared replicate IDs to
|
|
their source-ID lists. Aliased controls stay within a row; evaluation cannot
|
|
become new training replication. This pure function does NOT verify a file
|
|
receipt/seed derivation; the coordinator must independently do so.
|
|
"""
|
|
require(assay in ('normal', 'delayed') and mode in MODES and dan_type in DAN_TYPES and conditioned_cue in CUES,
|
|
'Unknown raw assay/mode/DAN/odor label')
|
|
require(isinstance(registered_training_streams, Mapping) and len(registered_training_streams) == 5,
|
|
'Require five predeclared independent training replicates')
|
|
registered, training_all = {}, set()
|
|
for identity, sources in registered_training_streams.items():
|
|
_token(identity, 'training replicate identity')
|
|
require(isinstance(sources, (tuple, list)) and len(sources) > 0, 'Missing actual training source identities')
|
|
values = tuple(_token(v, 'training source identity') for v in sources)
|
|
require(len(set(values)) == len(values) and not training_all.intersection(values),
|
|
'Training source reused across independent replicates')
|
|
training_all.update(values)
|
|
registered[identity] = values
|
|
require(isinstance(replicates, (tuple, list)) and len(replicates) == 5, 'Exactly five training rows required; do not count evaluation draws')
|
|
used_ids, used_eval, receipts, reports = set(), set(), set(), []
|
|
for row in replicates:
|
|
keys = {'replicate_id', 'replicate_unit', 'training_stream_ids', 'evaluation_stream_ids',
|
|
'assay', 'mode', 'dan_type', 'conditioned_cue', 'arms'}
|
|
require(isinstance(row, Mapping) and set(row) == keys, 'Invalid replicate schema')
|
|
identity = row['replicate_id']
|
|
require(identity in registered and identity not in used_ids and row['replicate_unit'] == 'independent_training',
|
|
'Unknown/duplicate training identity or evaluation pseudo-replication')
|
|
used_ids.add(identity)
|
|
require(isinstance(row['training_stream_ids'], (list, tuple)) and
|
|
tuple(row['training_stream_ids']) == registered[identity], 'Training lineage does not match preregistration')
|
|
require((row['assay'], row['mode'], row['dan_type'], row['conditioned_cue']) == (assay, mode, dan_type, conditioned_cue),
|
|
'Cannot pool modes, DAN branches, cue assignments or timing suites')
|
|
ev = row['evaluation_stream_ids']
|
|
require(isinstance(ev, (tuple, list)) and len(ev) > 0, 'Missing independent evaluation identity')
|
|
ev = tuple(_token(v, 'evaluation identity') for v in ev)
|
|
require(len(set(ev)) == len(ev) and not training_all.intersection(ev) and not used_eval.intersection(ev),
|
|
'Evaluation sources overlap training or another training replicate')
|
|
used_eval.update(ev)
|
|
require(isinstance(row['arms'], Mapping) and set(row['arms']) == set(ARMS), 'All six actual controls and paired arm are mandatory')
|
|
metrics = {}
|
|
for name in ARMS:
|
|
_execution(row['arms'][name], name, assay, receipts)
|
|
metrics[name] = _bound_metric(row['arms'][name]['metrics'], conditioned_cue)
|
|
baseline = metrics['paired']
|
|
# State baseline and populations must really be matched for each control.
|
|
for metric in metrics.values():
|
|
for cue in CUES:
|
|
a, b = baseline['cues'][cue], metric['cues'][cue]
|
|
require(a['membership'] == b['membership'] and a['nodes_raw_sha256'] == b['nodes_raw_sha256'] and
|
|
a['pre_state_raw_sha256'] == b['pre_state_raw_sha256'], 'Control population or pre-baseline mismatch')
|
|
s = baseline['rawS']
|
|
control_s = {name: metrics[name]['rawS'] for name in CONTROLS}
|
|
direction = s > 0 if dan_type == 'PAM15' else s < 0
|
|
beyond = {name: (s > value if dan_type == 'PAM15' else s < value) for name, value in control_s.items()}
|
|
neutral = baseline['neutral_cue']
|
|
c = baseline['cues'][conditioned_cue]['populations']['MBON']
|
|
n = baseline['cues'][neutral]['populations']['MBON']
|
|
functional = c['spike_count_l1'] > 0 or c['defined_first_latency_changed'] is True
|
|
predicates = dict(raw_signed_direction=bool(direction), beyond_each_actual_control=all(beyond.values()),
|
|
conditioned_final_g_exceeds_neutral=c['final_delta_g_max_abs'] > n['final_delta_g_max_abs'],
|
|
conditioned_count_or_defined_population_first_latency_change=bool(functional),
|
|
all_MBON_AB_euclidean_gain_positive=baseline['separation']['MBON']['gain'] > 0,
|
|
legacyDownstream_AB_euclidean_gain_positive=baseline['separation']['legacyDownstream']['gain'] > 0,
|
|
nonMBONDownstream_AB_euclidean_gain_positive=baseline['separation']['nonMBONDownstream']['gain'] > 0)
|
|
reports.append(dict(replicate_id=identity, rawS=s, raw_control_S=control_s,
|
|
population_separation_gains={name: baseline['separation'][name]['gain'] for name in SEPARATION_POPULATIONS},
|
|
beyond_controls= beyond, predicates=predicates, conjunction_pass=all(predicates.values()),
|
|
legacy_audit=dict(L_like_functional_with_explicit_missing_latency=bool(
|
|
predicates['conditioned_final_g_exceeds_neutral'] and functional),
|
|
H_like_MBON_positive_gain=predicates['all_MBON_AB_euclidean_gain_positive'],
|
|
H_like_downstream_population='legacyDownstream',
|
|
H_like_downstream_positive_gain=predicates['legacyDownstream_AB_euclidean_gain_positive'],
|
|
original_implementation_equivalence_claimed=False),
|
|
new_hypothesis_audit=dict(nonMBONDownstream_positive_gain=
|
|
predicates['nonMBONDownstream_AB_euclidean_gain_positive'],
|
|
inherited_from_old_H=False, either_downstream_may_rescue_other=False)))
|
|
reports.sort(key=lambda r: r['replicate_id'])
|
|
successful = sum(r['conjunction_pass'] for r in reports)
|
|
return dict(schema=SCHEMA, status='DRAFT-PROPOSAL-GATE-PASS' if successful >= 3 else 'DRAFT-PROPOSAL-GATE-FAIL',
|
|
proposal_gate_pass=successful >= 3, scientific_learning_pass_claimed=False, candidate_generated=False,
|
|
assay=assay, mode=mode, dan_type=dan_type, conditioned_cue=conditioned_cue,
|
|
required_training_replicates=5, successful_same_replicate_conjunctions=successful, required_majority=3,
|
|
actual_control_receipts=len(receipts) - 5, evaluation_draws_count_as_replicates=False,
|
|
execution_receipt_authority='caller must verify finalized receipts and seed registry; pure metrics do not read files',
|
|
historical_full_H_mean_control_rule_computed=False, replicates=reports)
|
|
|
|
|
|
def legacy_h_mean_control_audit(protocol_mbon_gains):
|
|
"""Old H scalar predicate only, kept separate from the new signed gate.
|
|
|
|
Input uses the ORIGINAL five protocol labels, each with five already-bound
|
|
training-replicate MBON Euclidean gains. The caller must explicitly define
|
|
any AN-to-old-protocol mapping; this function never picks a DAN/control or
|
|
pools AN cells. It is not the entire old H gate or a reproduced old result.
|
|
"""
|
|
names = ('appetitive-conditioning', 'aversive-conditioning', 'd-zero-control',
|
|
'random-reward-control', 'frozen-brain-control')
|
|
require(isinstance(protocol_mbon_gains, Mapping) and set(protocol_mbon_gains) == set(names),
|
|
'Legacy audit requires exactly the five original H protocol labels')
|
|
means = {}
|
|
for name in names:
|
|
values = protocol_mbon_gains[name]
|
|
require(isinstance(values, (list, tuple)) and len(values) == 5 and
|
|
all(type(v) in (int, float) and math.isfinite(v) for v in values),
|
|
'Legacy audit needs five finite, separately bound training-replicate gains')
|
|
means[name] = float(np.mean(values))
|
|
require(math.isfinite(means[name]), 'Nonfinite legacy mean')
|
|
floor, ceiling = min(means[names[0]], means[names[1]]), max(means[name] for name in names[2:])
|
|
return dict(schema=SCHEMA, audit='legacy_H_mean_control_predicate_only', protocol_means=means,
|
|
condition_floor=floor, control_ceiling=ceiling, predicate_pass=floor > ceiling,
|
|
source_reference='frozen phase4h.py:412-414', full_old_H_gate_computed=False,
|
|
scientific_learning_pass_claimed=False, AN_to_legacy_protocol_mapping_authority='caller; no automatic pooling')
|
|
|
|
|
|
def unavailable_assay(assay):
|
|
"""Do not turn a terminal delta or fresh reset into a retention/reversal PASS."""
|
|
require(assay in ('reversal', 'retention'), 'Unknown unavailable assay')
|
|
missing = {'reversal': ['bound acquisition/reversal/no-reversal lineage',
|
|
'original pre baseline and original cue orientation', 'terminal reversal count and both cue event outputs'],
|
|
'retention': ['validated continuous-state checkpoint/queue/RNG schema',
|
|
'bound same idle trajectory at 1 s and 10 s', 'matched frozen continuation and non-feedback probes']}
|
|
return dict(schema=SCHEMA, assay=assay, status='NOT-IMPLEMENTED', proposal_gate_pass=None,
|
|
scientific_learning_pass_claimed=False, candidate_generated=False, missing=missing[assay])
|