Digital Drosophila Archive
146 lines
9.6 KiB
Python
146 lines
9.6 KiB
Python
"""Register the exact zero/static interface tests; create no brain trajectory."""
|
|
import ast
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import platform
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import numpy as np
|
|
import pandas as pd
|
|
from context import (ROOT, MODEL, EXP, AK, AL, H, AG, META, AJ_MODEL, CFG, PINS,
|
|
read, save, digest, sources, integrity, write_npz)
|
|
|
|
|
|
def old_base_candidate(raw, source_path):
|
|
"""Evaluate exactly the pinned old assignment RHS; all candidate KC signs +1."""
|
|
assert raw.dtype == np.int64 and raw.ndim == 1 and np.all(raw > 0)
|
|
tree = ast.parse(source_path.read_text())
|
|
runtime = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == 'Runtime')
|
|
init = next(n for n in runtime.body if isinstance(n, ast.FunctionDef) and n.name == '__init__')
|
|
selected = [n for n in ast.walk(init) if isinstance(n, ast.Assign) and any(
|
|
isinstance(t, ast.Subscript) and isinstance(t.value, ast.Name) and t.value.id == 'effective'
|
|
for t in n.targets)]
|
|
assert len(selected) == 1
|
|
value = selected[0].value
|
|
assert ast.unparse(value) == "raw_np[a:b].astype(np.float32) * signs[src] * float(cfg['model']['w_syn_mv_per_anatomical_synapse'])"
|
|
env = dict(raw_np=raw, a=0, b=len(raw), np=np, signs=np.array([1], np.int8), src=0,
|
|
cfg={'model': {'w_syn_mv_per_anatomical_synapse': .275}}, float=float)
|
|
out = eval(compile(ast.Expression(value), str(source_path), 'eval'), {'__builtins__': {}}, env)
|
|
assert out.dtype == np.float32 and np.isfinite(out).all() and np.all(out > 0)
|
|
return out, hashlib.sha256(ast.dump(value, include_attributes=False).encode()).hexdigest()
|
|
|
|
|
|
def main():
|
|
assert not EXP.exists() or (EXP.is_dir() and not any(EXP.iterdir())), 'Never reuse a populated experiment'
|
|
assert shutil.disk_usage(EXP.parent).free > CFG['output_budget_bytes'] + 10 * 1024**3
|
|
for relative, sha in PINS.items():
|
|
assert digest(ROOT / relative) == sha, relative
|
|
assert read(AL / 'audit/final-status.json')['status'] == 'PASS-CONDITIONAL-CREDIT-TRANSFER-AUDIT-NOT-LEARNING'
|
|
assert CFG['stages'] == ['zero', 'static25'] and not CFG['training'] and not CFG['candidate_generation']
|
|
records = read(AK / 'protocol/validation-cases.json')
|
|
assert len(records) == 120 and len({r['name'] for r in records}) == 120
|
|
first_seed = min(r['seed'] for r in records)
|
|
planned = [dict(row, repeat=repeat) for row in records
|
|
for repeat in ((0, 1) if row['seed'] == first_seed else (0,))]
|
|
assert len(planned) == 144
|
|
inputs = dict(PINS)
|
|
static = [AK / 'protocol/validation-cases.json',
|
|
H / 'mappings/mbon-to-downstream-actual-edges.parquet',
|
|
ROOT / 'experiments/phase4l-olfactory-grounding/odor-encoder/selection.json',
|
|
ROOT / 'experiments/phase4af-author-fullgraph/oracle/no_input/coefficients.json']
|
|
static += list((AG / 'graphs/full').glob('*.npy')) + list((AG / 'apl-bundle').glob('*.npy'))
|
|
raw_path = ROOT / 'datasets/male-cns-v1.0/processed/neuron-only/gpu/csr-weights.int64.npy'
|
|
static.append(raw_path)
|
|
for row in planned:
|
|
case = AK / 'cases' / row['cohort'] / row['name']
|
|
spec = read(case / 'spec.json')
|
|
assert spec['steps'] == 1000 and spec['dt_ms'] == .1 and spec['apl_delay_steps'] == 18
|
|
assert spec['model']['w_syn_mv_per_anatomical_synapse'] == .275
|
|
static += [case / 'spec.json', case / 'requests.npz']
|
|
for p in static:
|
|
inputs[p.relative_to(ROOT).as_posix()] = digest(p)
|
|
for name in ('protocol', 'overlays', 'gpu', 'audit', 'regression', 'logs', 'extension-build', 'summaries'):
|
|
(EXP / name).mkdir(parents=True, exist_ok=False)
|
|
test = subprocess.run([sys.executable, '-B', '-m', 'unittest', 'discover', '-v', '-s', str(MODEL)],
|
|
cwd=MODEL, text=True, capture_output=True)
|
|
with (EXP / 'regression/unit-tests.log').open('x', encoding='utf-8') as f:
|
|
f.write(test.stdout + test.stderr)
|
|
save(EXP / 'regression/unit-tests.json', dict(pass_all=test.returncode == 0,
|
|
exit_code=test.returncode, log_sha256=digest(EXP / 'regression/unit-tests.log')))
|
|
assert test.returncode == 0, 'Unit tests failed; preserve partial experiment'
|
|
assert 'skipped=' not in test.stderr, 'All server compatibility tests must execute'
|
|
save(EXP / 'protocol/preregistered.json', dict(timestamp_utc=datetime.now(timezone.utc).isoformat(),
|
|
sources=sources(), config=CFG, planned=planned,
|
|
inputs=[dict(path=k, sha256=v) for k, v in sorted(inputs.items())],
|
|
parent_commit=subprocess.check_output(['git', '-C', str(ROOT), 'rev-parse', 'HEAD'], text=True).strip(),
|
|
before_new_brain_trajectories=True, prior_AK_AL_results_known=True,
|
|
no_dynamic_pair_selection=True, training=False, candidate_generated=False,
|
|
python=sys.version, numpy=np.__version__, platform=platform.platform()))
|
|
save(EXP / 'regression/old-integrity-before.json', integrity())
|
|
candidate = pd.read_parquet(H / 'plastic-mask/plastic-edge-candidates.parquet')
|
|
mask_path = H / 'plastic-mask/plastic-edge-ids.int64.npy'
|
|
ids = np.load(mask_path, allow_pickle=False)
|
|
mask = np.load(H / 'plastic-mask/plastic-edge-mask.bool.npy', allow_pickle=False)
|
|
assert ids.dtype == np.int64 and ids.shape == (14232,)
|
|
assert np.array_equal(candidate.edge_id, ids) and np.array_equal(np.flatnonzero(mask), ids)
|
|
graph = {k: np.load(AG / 'graphs/full' / (k + '.npy'), mmap_mode='r')
|
|
for k in ('src', 'dst', 'edge_id', 'weights', 'signed_raw', 'nodes')}
|
|
assert np.array_equal(graph['nodes'], np.arange(166700, dtype=np.int32))
|
|
assert np.array_equal(graph['edge_id'], np.arange(25582938, dtype=np.int64))
|
|
assert np.array_equal(graph['src'][ids], candidate.source_neuron_idx)
|
|
assert np.array_equal(graph['dst'][ids], candidate.target_neuron_idx)
|
|
raw = np.load(raw_path, mmap_mode='r')
|
|
assert np.array_equal(raw[ids], candidate.raw_synapse_count)
|
|
assert np.array_equal(graph['signed_raw'][ids], raw[ids]), 'Original candidate sign must be +1'
|
|
assert np.all(graph['weights'][ids] > 0) and (candidate.plasticity_polarity == -1).all()
|
|
base, rhs_sha = old_base_candidate(raw[ids], ROOT / 'models/male-cns-lif/phase4c.py')
|
|
base_path = EXP / 'protocol/base-candidate.float32-mV.npy'
|
|
with base_path.open('xb') as f:
|
|
np.save(f, base, allow_pickle=False)
|
|
for stage in CFG['stages']:
|
|
delta = np.zeros(len(ids), np.float32) if stage == 'zero' else (-float(.25) * base).astype(np.float32)
|
|
path = EXP / 'overlays' / (stage + '.npz')
|
|
write_npz(path, edge_id=ids, plastic_delta_mv=delta)
|
|
save(EXP / 'overlays' / (stage + '.json'), dict(schema='phase4am-overlay-v1', overlay_name=stage,
|
|
overlay_sha256=digest(path), mask_path=str(mask_path), mask_sha256=digest(mask_path),
|
|
base_candidate_path=str(base_path), base_candidate_sha256=digest(base_path),
|
|
aj_model_dir=str(AJ_MODEL), base_rhs_ast_sha256=rhs_sha,
|
|
old_f32_expression_source_sha256=digest(ROOT / 'models/male-cns-lif/phase4c.py'),
|
|
training=False, fixed_for_whole_trial=True))
|
|
meta = pd.read_parquet(META).sort_values('neuron_idx').reset_index(drop=True)
|
|
assert np.array_equal(meta.neuron_idx.to_numpy(), graph['nodes'])
|
|
kc = np.flatnonzero(meta.annotation_class.eq('Kenyon_Cell')).astype(np.int32)
|
|
mbon = np.flatnonzero(meta.annotation_class.eq('MBON')).astype(np.int32)
|
|
assert len(kc) == 4064 and len(mbon) == 97
|
|
assert np.isin(candidate.source_neuron_idx, kc).all() and np.isin(candidate.target_neuron_idx, mbon).all()
|
|
down = pd.read_parquet(H / 'mappings/mbon-to-downstream-actual-edges.parquet')
|
|
assert np.array_equal(graph['src'][down.edge_id.to_numpy()], down.source_neuron_idx.to_numpy())
|
|
assert np.array_equal(graph['dst'][down.edge_id.to_numpy()], down.target_neuron_idx.to_numpy())
|
|
populations = dict(KC=kc, MBON=mbon,
|
|
mask_recipient_MBON=np.unique(candidate.target_neuron_idx.to_numpy(dtype=np.int32)),
|
|
DAN=np.flatnonzero(meta.annotation_class.eq('DAN')).astype(np.int32),
|
|
MBON_downstream=np.unique(down.target_neuron_idx.to_numpy(dtype=np.int32)))
|
|
# Exact observed superclass strings are checked; no inferred body identity.
|
|
role_labels = {'descending': ['descending_neuron', 'descending_neuron_tbc'],
|
|
'motor': ['vnc_motor', 'cb_motor']}
|
|
for name, labels in role_labels.items():
|
|
assert set(labels).issubset(set(meta.annotation_superclass.dropna().astype(str))), labels
|
|
populations[name] = np.flatnonzero(meta.annotation_superclass.isin(labels)).astype(np.int32)
|
|
write_npz(EXP / 'protocol/populations.npz', **populations)
|
|
save(EXP / 'protocol/static-audit.json', dict(candidate_edges=len(ids), mask_fixed=True,
|
|
old_base_rhs_ast_sha256=rhs_sha, populations={k: len(v) for k, v in populations.items()},
|
|
exact_descending_motor_superclass_labels=role_labels,
|
|
compartments={str(k): len(g) for k, g in candidate.groupby('target_compartment_verbatim')},
|
|
old_selected_odors=read(ROOT / 'experiments/phase4l-olfactory-grounding/odor-encoder/selection.json'),
|
|
source_mapping='Exact annotation_class/superclass and frozen actual-edge table; no inferred type/body IDs',
|
|
old_base_signed_positive=True, training=False))
|
|
save(EXP / 'protocol/prepared-hashes.json', [dict(path=p.relative_to(EXP).as_posix(), sha256=digest(p))
|
|
for folder in ('protocol', 'overlays') for p in sorted((EXP / folder).rglob('*'))
|
|
if p.is_file() and p.name != 'prepared-hashes.json'])
|
|
print('PREREGISTERED zero 144 then static25 144; no brain trajectories or training', flush=True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|