Digital Drosophila Archive
289 lines
19 KiB
Python
289 lines
19 KiB
Python
"""UNREGISTERED bounded preflight input planning, not a conditioning runner.
|
|
|
|
Only the explicitly called generate_epoch_inputs performs random sampling; it
|
|
delegates to conditioning_inputs' single full-shape CUDA draw. The plan and
|
|
static rate loader do no sampling, neural execution, plasticity, or I/O writes.
|
|
The coordinator owns registration, full seals, graph/overlay authority, device
|
|
UUID admission, and every epoch update/checkpoint receipt.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import copy
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
|
|
import conditioning_inputs as inputs
|
|
import trial_plan
|
|
|
|
|
|
MASTER = 20260930
|
|
BLOCK = 'physiological__PAM15__reinforce_A__rep01'
|
|
SCHEMA = 'phase4an-conditioning-preflight-inputs-v1-unregistered'
|
|
STATIC = 'experiments/phase4l-olfactory-grounding/'
|
|
SOURCE_REL = 'models/male-cns-lif/phase4l.py'
|
|
CONFIG_REL = 'models/male-cns-lif/olfactory-grounding-v1/phase4l.yaml'
|
|
SOURCE_PINS = {
|
|
SOURCE_REL: 'e97d58d8c1f44920d0e953f8f5831ddc86fa0198784e86460e315de4116ac553',
|
|
CONFIG_REL: '9b8bcb7e031bf9063033a1854a439a7cd0bf6730301c2d2d02ca99a76f098d6e',
|
|
STATIC + 'odor-encoder/frozen-odor-panel.parquet': '907f16a56e024a8e2112b5751e53b0e6037b329fa110098af3d31b119c10a1a0',
|
|
STATIC + 'odor-encoder/mapped-glomerular-vectors.npz': 'f057563152e0fa22060080af13019f319e7f36563a775eff829bbc1929c451b0',
|
|
STATIC + 'odor-encoder/selection.json': '1b832becf9d08935716fe9ba5e3cd3e619e7d208dfb43c4fa197c969d095f113',
|
|
STATIC + 'receptor-mapping/olfactory-mapping.parquet': 'a01d27000533e9844458a4c559cac084f1c34ca20a1104e97e6a0f72b0cc93b1',
|
|
}
|
|
|
|
|
|
def require(ok, message):
|
|
if not ok:
|
|
raise ValueError(message)
|
|
|
|
|
|
def _canonical(value):
|
|
return json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=False, allow_nan=False)
|
|
|
|
|
|
def _sha(value):
|
|
return hashlib.sha256(_canonical(value).encode('utf-8')).hexdigest()
|
|
|
|
|
|
def build_preflight_plan():
|
|
"""Rebuild the whole semantic plan, then select by rule, never outcomes.
|
|
|
|
D1/D0/frozen share the two paired input epochs. Every neural epoch starts
|
|
fresh; only the externally verified plastic delta can cross its boundary.
|
|
This inventory does not enable the complete 68,000-epoch protocol.
|
|
"""
|
|
full = trial_plan.build_trial_plan(MASTER)
|
|
trial_plan.validate_trial_plan(full)
|
|
block = full['plan']['blocks'][0]
|
|
require(block['block_id'] == BLOCK and block['mode'] == 'physiological' and
|
|
block['dan_type'] == 'PAM15' and block['compartment_verbatim'] == "y5B'2a", 'First fixed block changed')
|
|
suite = next(s for s in block['suites'] if s['assay'] == 'normal')
|
|
paired = next(a for a in suite['arms'] if a['arm'] == 'paired')
|
|
first = next(i for i, epoch in enumerate(paired['epochs']) if epoch['us']['requested'])
|
|
require(first + 1 < len(paired['epochs']), 'Requested US lacks its following complete epoch')
|
|
epochs = copy.deepcopy(paired['epochs'][first:first + 2])
|
|
calls = []
|
|
for arm in ('D1', 'D0', 'frozen'):
|
|
for mode in ('reference', 'light'):
|
|
previous = None
|
|
for position, epoch in enumerate(epochs):
|
|
_validate_epoch(epoch)
|
|
row_id = f'{arm}--{mode}--epoch-{position:03d}'
|
|
calls.append(dict(row_id=row_id, arm=arm, diagnostic_mode=mode,
|
|
input_epoch_index=position, epoch_id=epoch['epoch_id'],
|
|
full_plan_epoch_index=epoch['epoch_index'], steps=1200,
|
|
d_command=0.0 if arm == 'D0' else epoch['D_command'],
|
|
frozen=arm == 'frozen', previous_delta_row_id=previous,
|
|
neural_state='fresh_rest_zero_ref_zero_spike_queue_zero_APL_queue',
|
|
input_sharing='same saved CS and US arrays across arms and diagnostic modes'))
|
|
previous = row_id
|
|
result = dict(schema=SCHEMA, master=MASTER, block_id=BLOCK, mode='physiological',
|
|
dan_type='PAM15', compartment_verbatim="y5B'2a", full_trial_plan=full,
|
|
selected_epochs=epochs, calls=calls, total_calls=12, total_ticks=14400,
|
|
input_generation_calls=2, input_generation_executed=False, brain_runs_executed=0,
|
|
complete_conditioning_authorized=False, training_result_claimed=False,
|
|
selection_rule='first normal paired requested US epoch in first fixed block, plus its immediate successor')
|
|
result['payload_sha256'] = _sha(result)
|
|
return result
|
|
|
|
|
|
def validate_preflight_plan(value):
|
|
require(type(value) is dict and _canonical(value) == _canonical(build_preflight_plan()),
|
|
'Preflight inventory differs from full deterministic reconstruction')
|
|
return True
|
|
|
|
|
|
def _validate_epoch(epoch):
|
|
require(type(epoch) is dict and epoch.get('steps') == 1200 and epoch.get('duration_ms') == 120 and
|
|
epoch.get('initial_neural_state') == 'frozen_reset_including_KC_trace_and_both_delay_queues',
|
|
'Only complete fresh normal 120 ms epochs are in this preflight')
|
|
cs, us = epoch.get('cs'), epoch.get('us')
|
|
require(type(cs) is dict and cs.get('cue') in ('A', 'B') and cs.get('odor') == trial_plan.ODORS[cs['cue']] and
|
|
type(cs.get('seed')) is int and 0 <= cs['seed'] < 2**63 and cs.get('seed_bits') == 63 and
|
|
cs.get('start_tick') == 0 and cs.get('stop_tick_exclusive') == 1000, 'CS semantic descriptor changed')
|
|
require(type(us) is dict and type(us.get('requested')) is bool and us.get('nominal_slot_tick') == 900 and
|
|
us.get('rng_seed') is None and us.get('rng_draws') == 0, 'US must remain a deterministic scheduled pulse')
|
|
require((us['requested'] and us.get('tick') == 900 and us.get('request_rule') == 'fixed_single_pulse_all_bound_DAN_targets') or
|
|
(not us['requested'] and us.get('tick') is None and us.get('request_rule') == 'none'), 'US timing/request rule changed')
|
|
|
|
|
|
def _extract_drive(raw, filename):
|
|
"""Compile only the SHA-pinned pure pn_drive AST; never import old L."""
|
|
import pandas as pd
|
|
require(hashlib.sha256(raw).hexdigest() == SOURCE_PINS[SOURCE_REL], 'Frozen L source SHA mismatch')
|
|
nodes = [n for n in ast.parse(raw, filename=filename).body if isinstance(n, ast.FunctionDef) and n.name == 'pn_drive']
|
|
require(len(nodes) == 1 and not nodes[0].decorator_list, 'Unique undecorated pn_drive required')
|
|
namespace = {'np': np, 'pd': pd, 'Any': Any}
|
|
exec(compile(ast.Module(body=nodes, type_ignores=[]), filename, 'exec'), namespace)
|
|
return namespace['pn_drive'], hashlib.sha256(ast.dump(nodes[0], include_attributes=False).encode()).hexdigest()
|
|
|
|
|
|
def _rate_tables(selection, panel, vectors, mapping, config, drive):
|
|
"""Pure static-table derivation, preserving every original PN column."""
|
|
require(selection.get('selected_before_KC_simulation') is True and len(panel) == 12 and
|
|
panel.odor_id.is_unique and panel.split.tolist() == ['calibration'] * 6 + ['heldout'] * 6,
|
|
'Frozen static odor panel schema changed')
|
|
gloms = selection['glomeruli_order']
|
|
require(type(gloms) is list and len(gloms) == len(set(gloms)) and len(gloms) > 0 and
|
|
set(vectors) == set(panel.odor_id), 'Frozen vector/glomerulus inventory mismatch')
|
|
encoder = config['odor_encoder']
|
|
require(encoder['max_total_rate_per_glomerulus_hz'] == 400.0 and
|
|
encoder['modes'] == ['physiological', 'matched_energy'], 'Frozen rate transform changed')
|
|
median = float(np.median(panel.mapped_response_sum))
|
|
require(np.isfinite(median) and median > 0, 'Invalid panel response median')
|
|
tables = {}
|
|
for cue, name in trial_plan.ODORS.items():
|
|
selected = selection['ODOR_' + cue]
|
|
require(selected['odor_name'] == name, 'Selected frozen A/B odor name changed')
|
|
odor_id = selected['odor_id']
|
|
positions = np.flatnonzero(panel.odor_id.to_numpy() == odor_id)
|
|
require(len(positions) == 1, 'Selected odor ID not unique in frozen panel')
|
|
panel_row = panel.iloc[int(positions[0])]
|
|
require(panel_row.odor_name == name and panel_row.formal_label == 'ODOR_' + cue, 'Panel/selection label mismatch')
|
|
vector = vectors[odor_id]
|
|
require(isinstance(vector, np.ndarray) and vector.dtype == np.float32 and vector.shape == (len(gloms),) and
|
|
np.isfinite(vector).all() and np.all(vector >= 0), 'Invalid frozen glomerular vector; never clip')
|
|
targets, rates, frame = drive(odor_id, vector, mapping, gloms, 'physiological', median, config)
|
|
inputs.probabilities_f32(targets, rates)
|
|
require(len(targets) > 0 and np.all(rates > 0), 'Empty or nonpositive selected PN drive')
|
|
require(frame.neuron_idx.to_numpy(np.int32).tobytes() == targets.tobytes() and
|
|
frame.rate_hz.to_numpy(np.float32).tobytes() == rates.tobytes(), 'Original frame/array ordering mismatch')
|
|
targets, rates = targets.copy(), rates.copy()
|
|
targets.setflags(write=False); rates.setflags(write=False)
|
|
tables[cue] = dict(cue=cue, odor_id=str(odor_id), odor_name=name, mode='physiological',
|
|
source_panel_row=int(positions[0]), source_panel_split=str(panel_row.split),
|
|
targets=targets, rates_hz=rates, identities={
|
|
'targets': inputs.array_identity(targets), 'rates_hz': inputs.array_identity(rates)},
|
|
glomeruli_order=copy.deepcopy(gloms), mapped_vector_identity=inputs.array_identity(vector))
|
|
return tables, median
|
|
|
|
|
|
def load_rate_tables(root):
|
|
"""Read only the six exact static/source files pinned by frozen AK.
|
|
|
|
This does not read any historical neural result or rerun odor selection.
|
|
Source files are checked before parsing; the parsed bytes are the same
|
|
checked snapshots. Full prior seal/registration authority remains external.
|
|
"""
|
|
import pandas as pd
|
|
import yaml
|
|
root = Path(root)
|
|
raw = {rel: (root / rel).read_bytes() for rel in SOURCE_PINS}
|
|
require(all(hashlib.sha256(value).hexdigest() == SOURCE_PINS[rel] for rel, value in raw.items()),
|
|
'Frozen static encoder/source SHA mismatch')
|
|
source, ast_sha = _extract_drive(raw[SOURCE_REL], str(root / SOURCE_REL))
|
|
selection = json.loads(raw[STATIC + 'odor-encoder/selection.json'])
|
|
panel = pd.read_parquet(io.BytesIO(raw[STATIC + 'odor-encoder/frozen-odor-panel.parquet']))
|
|
mapping = pd.read_parquet(io.BytesIO(raw[STATIC + 'receptor-mapping/olfactory-mapping.parquet']))
|
|
config = yaml.safe_load(raw[CONFIG_REL])
|
|
with np.load(io.BytesIO(raw[STATIC + 'odor-encoder/mapped-glomerular-vectors.npz']), allow_pickle=False) as values:
|
|
require(set(values.files) == set(panel.odor_id) | {'glomeruli'} and
|
|
values['glomeruli'].tolist() == selection['glomeruli_order'], 'Frozen NPZ inventory/order mismatch')
|
|
vectors = {str(oid): values[oid].copy() for oid in panel.odor_id}
|
|
tables, median = _rate_tables(selection, panel, vectors, mapping, config, source)
|
|
return dict(tables=tables, audit=dict(schema=SCHEMA, static_sources=copy.deepcopy(SOURCE_PINS),
|
|
original_pn_drive_AST_sha256=ast_sha, historical_dynamic_results_read=False,
|
|
original_function_called_without_PN_reordering=True, mode='physiological',
|
|
median_mapped_response_sum=median, full_seal_verification_owned_by_coordinator=True,
|
|
random_draws=0, brain_runs=0, numpy_version=np.__version__, pandas_version=pd.__version__))
|
|
|
|
|
|
def _validate_table(table, cue):
|
|
require(type(table) is dict and table.get('cue') == cue and table.get('odor_name') == trial_plan.ODORS[cue] and
|
|
table.get('mode') == 'physiological' and type(table.get('odor_id')) is str and table['odor_id'], 'Rate-table cue identity mismatch')
|
|
inputs.probabilities_f32(table['targets'], table['rates_hz'])
|
|
require(table['identities'] == {key: inputs.array_identity(table[key]) for key in ('targets', 'rates_hz')},
|
|
'Rate-table arrays changed after authority binding')
|
|
|
|
|
|
def generate_epoch_inputs(epoch, tables, dan_targets, *, device, max_uniform_draws, conflict_policy='error'):
|
|
"""Future explicit sampler, called once per selected input epoch, not arm.
|
|
|
|
No brain reset is called. The scheduled US request is identical for D1,
|
|
D0, and frozen; D-command and update policy belong to the epoch authority.
|
|
The caller must preserve and share these returned bytes across all arms.
|
|
"""
|
|
_validate_epoch(epoch)
|
|
cs = epoch['cs']; table = tables[cs['cue']]; _validate_table(table, cs['cue'])
|
|
require(isinstance(dan_targets, np.ndarray) and dan_targets.dtype == np.int32 and dan_targets.ndim == 1 and
|
|
len(dan_targets) > 0 and np.all((dan_targets >= 0) & (dan_targets < inputs.N_NEURONS)) and
|
|
(len(dan_targets) < 2 or np.all(dan_targets[1:] > dan_targets[:-1])), 'Externally bound DAN globals must be sorted unique int32')
|
|
require(conflict_policy in ('error', 'boolean_union'), 'Unknown explicit request composition policy')
|
|
cs_result = inputs.generate_cs_requests(table['targets'], table['rates_hz'], seed=cs['seed'], steps=1200,
|
|
cs_start_tick=0, cs_stop_tick=1000, device=device, max_uniform_draws=max_uniform_draws)
|
|
nodes = dan_targets.copy() if epoch['us']['requested'] else np.empty(0, np.int32)
|
|
us = {'tick': np.full(len(nodes), 900, np.int32), 'neuron_idx': nodes}
|
|
for value in us.values(): value.setflags(write=False)
|
|
composed = inputs.merge_requests(cs_result['requests'], us, steps=1200, conflict_policy=conflict_policy)
|
|
return dict(requests=composed['requests'], cs_requests=cs_result['requests'], us_requests=us,
|
|
source_mask=composed['source_mask'], audit=dict(schema=SCHEMA, epoch_id=epoch['epoch_id'],
|
|
CS_sampler=cs_result['audit'], composition=composed['audit'],
|
|
DAN_targets_identity=inputs.array_identity(dan_targets), requested_US=epoch['us']['requested'],
|
|
scheduled_US_tick=epoch['us']['tick'], CS_shared_across_arms=True, US_shared_across_arms=True,
|
|
US_rng_draws=0, semantic_seed_never_passed_to_brain_reset=True, brain_reset_called=False,
|
|
source_authority_and_saved_array_rehash_owned_by_coordinator=True))
|
|
|
|
|
|
def derive_case_spec(original, row, epoch, rate_table, *, brain_reset_seed, budget_root, output_budget_bytes):
|
|
"""Copy an externally authenticated wholegraph spec with explicit changes.
|
|
|
|
``seed`` is an independent conservative 31-bit reset-control value, not a
|
|
conversion of the semantic input seed. The inherited deterministic AJ
|
|
runtime consumes saved requests, not this input RNG. No checkpoint carry
|
|
is admitted: the next epoch carries only its separately verified delta.
|
|
"""
|
|
_validate_epoch(epoch); _validate_table(rate_table, epoch['cs']['cue'])
|
|
require(type(original) is dict and original.get('kind') == 'fullgraph' and original.get('dt_ms') == .1 and
|
|
original.get('apl_delay_steps') == 18 and original.get('precision') == 'float64' and
|
|
original.get('apl_precision') == 'float32' and original.get('record_history') is False and
|
|
original.get('within_trial_weight_updates', False) is False, 'Anchored fresh D18 fullgraph spec required')
|
|
require(not any('checkpoint' in str(k).lower() or k in ('initial_state', 'resume', 'continuation', 'global_tick_offset') for k in original),
|
|
'No neural checkpoint continuation in this two-epoch preflight')
|
|
require(type(brain_reset_seed) is int and 0 <= brain_reset_seed < 2**31, 'Independent conservative 31-bit brain reset seed required')
|
|
require(type(row) is dict and row.get('arm') in ('D1', 'D0', 'frozen') and row.get('diagnostic_mode') in ('reference', 'light') and
|
|
row.get('epoch_id') == epoch['epoch_id'] and row.get('steps') == 1200 and
|
|
row.get('frozen') is (row['arm'] == 'frozen') and
|
|
row.get('d_command') == (0.0 if row['arm'] == 'D0' else epoch['D_command']), 'Row/epoch command mismatch')
|
|
require(type(row.get('input_epoch_index')) is int and row['input_epoch_index'] in (0, 1) and
|
|
row.get('row_id') == f"{row['arm']}--{row['diagnostic_mode']}--epoch-{row['input_epoch_index']:03d}", 'Unsafe row identity')
|
|
require(type(budget_root) is str and budget_root.startswith('/') and not budget_root.startswith('//') and
|
|
'..' not in PurePosixPath(budget_root).parts and
|
|
type(output_budget_bytes) is int and output_budget_bytes > 0, 'Explicit absolute output budget required')
|
|
old = PurePosixPath(original['budget_root']); new = PurePosixPath(budget_root)
|
|
require(old.is_absolute() and not new.is_relative_to(old) and not old.is_relative_to(new), 'Budget cannot overlap old experiment')
|
|
result = copy.deepcopy(original)
|
|
removed = {key: result.pop(key) for key in ('validation_cohort', 'odor_index', 'expected_external_events', 'training') if key in result}
|
|
cs = epoch['cs']; rates = rate_table['rates_hz']
|
|
result.update(name=row['row_id'], steps=1200, duration_ms=120.0, max_steps=1200,
|
|
budget_root=budget_root, output_budget_bytes=output_budget_bytes, seed=brain_reset_seed,
|
|
cs_input_seed=cs['seed'], cs_input_seed_bits=63,
|
|
seed_scope='seed is separate reset control; full cs_input_seed used only by saved-input sampler',
|
|
mode='physiological', odor_id=rate_table['odor_id'], odor_name=rate_table['odor_name'],
|
|
source_panel_row=rate_table['source_panel_row'], source_panel_split=rate_table['source_panel_split'],
|
|
rate_sum_hz=float(rates.sum()), target_count=len(rate_table['targets']),
|
|
expected_CS_requests=float(np.sum(inputs.probabilities_f32(rate_table['targets'], rates), dtype=np.float64) * 1000),
|
|
CS_window=[0, 1000], within_trial_weight_updates=False,
|
|
outer_protocol='conditioning_interface_preflight_not_full_conditioning',
|
|
outer_epoch_id=epoch['epoch_id'], outer_arm=row['arm'])
|
|
changes = {k: dict(present_before=k in original, before=copy.deepcopy(original.get(k)), after=copy.deepcopy(v))
|
|
for k, v in result.items() if k not in original or _canonical(v) != _canonical(original[k])}
|
|
allowed = {'name', 'steps', 'duration_ms', 'max_steps', 'budget_root', 'output_budget_bytes', 'seed',
|
|
'cs_input_seed', 'cs_input_seed_bits', 'seed_scope', 'mode', 'odor_id', 'odor_name', 'source_panel_row',
|
|
'source_panel_split', 'rate_sum_hz', 'target_count', 'expected_CS_requests', 'CS_window',
|
|
'within_trial_weight_updates', 'outer_protocol', 'outer_epoch_id', 'outer_arm'}
|
|
require(set(changes) <= allowed and all(_canonical(result[k]) == _canonical(v) for k, v in original.items()
|
|
if k not in allowed and k not in removed), 'Unexpected frozen spec change')
|
|
return result, dict(schema=SCHEMA, explicit_field_changes=changes, removed_stale_input_metadata=removed,
|
|
original_spec_sha256=_sha(original), derived_spec_sha256=_sha(result),
|
|
graph_APL_model_coefficients_and_other_fields_unchanged=True,
|
|
independent_brain_reset_seed=brain_reset_seed, semantic_input_seed=cs['seed'],
|
|
neural_state_between_epochs='fresh reset; only plastic delta may carry via separate verified authority',
|
|
D_and_update_commands_owned_by_epoch_authority=True, random_draws=0, brain_runs=0,
|
|
outer_learning_pass_claimed=False, full_conditioning_authorized=False)
|