"""Full-connectome sensory-motor feedback, without an external behavioral policy.""" from pathlib import Path import os,sys,json,hashlib,time,platform,re,ast,resource,gc,importlib.metadata from datetime import datetime,timezone import numpy as np import pandas as pd import brian2 as b ROOT=Path(__file__).resolve().parent;P=json.loads((ROOT/'protocol.json').read_text()) BASE=Path('/home/hskim/ops/dropkit-fly/runs/2026-10-26/2026-10-26_fly-baseline_v01') assert platform.system()=='Linux' and P['host']=='linuxpc' OUT=ROOT/'results';OUT.mkdir();CACHE=ROOT/'cache';CACHE.mkdir() b.prefs.codegen.target='cython';b.prefs.codegen.runtime.cython.cache_dir=str(CACHE) assert float(b.defaultclock.dt/b.ms)==P['dt_ms'] sys.path.insert(0,str(BASE/'source'));import model def sha(path):return hashlib.sha256(path.read_bytes()).hexdigest() def original_blobs(): manifest=json.loads((BASE/'source-manifest.json').read_text());assert manifest['commit']==P['source_commit'];checks=[] for row in manifest['files']: path=BASE/'source'/row['name'];h=hashlib.sha1();h.update(f'blob {path.stat().st_size}\0'.encode()) with path.open('rb')as f: for chunk in iter(lambda:f.read(8*1024*1024),b''):h.update(chunk) checks.append({'name':row['name'],'gitBlob':h.hexdigest(),'match':h.hexdigest()==row['sha']}) assert len(checks)==7 and all(c['match']for c in checks);return checks before=original_blobs();notebook=json.loads((BASE/'source/example.ipynb').read_text()) source='\n'.join(''.join(c.get('source',[]))for c in notebook['cells']) inputs=ast.literal_eval(re.search(r'neu_sugar\s*=\s*(\[[^\]]+\])',source,re.S).group(1));assert len(inputs)==21 comp=pd.read_csv(BASE/'source/2023_03_23_completeness_630_final.csv',index_col=0);ids=np.asarray(comp.index,dtype=np.int64) f2i={int(v):i for i,v in enumerate(ids)};exc=[f2i[v]for v in inputs];mn9=f2i[P['motor_id']];assert mn9 not in exc trials=[];began=time.monotonic();started=datetime.now(timezone.utc).isoformat() for seed in P['seeds']: rng=np.random.default_rng(seed);steps,channels=np.nonzero(rng.random((10000,21))0 if condition in ['mn9-input-cut','mn9-restored']:syn.w[incoming]=0*b.volt if condition=='mn9-restored':syn.w[incoming]=original[incoming]*b.volt expected=original.copy() if condition=='mn9-input-cut':expected[incoming]=0 assert np.array_equal(np.asarray(syn.w_[:]),expected) weights_hash=hashlib.sha256(expected.tobytes()).hexdigest() driver=b.SpikeGeneratorGroup(21,channels,steps*P['dt_ms']*b.ms,name='candidate_sugar_drive') inject=b.Synapses(driver,neu,'gate: 1 (shared)',on_pre='v_post += gate * input_weight',namespace={'input_weight':model.default_params['w_syn']*model.default_params['f_poi']},name='contact_sugar_encoder') inject.connect(i=np.arange(21),j=exc);input_mon=b.SpikeMonitor(driver) net=b.Network(neu,syn,mon,driver,inject,input_mon) remaining=P['resource_units'];previous_count=0;feedback=[] for window in range(P['duration_ms']//P['feedback_ms']): gate=int(remaining>0 and condition!='sensory-block');inject.gate=gate resource_before=remaining;net.run(P['feedback_ms']*b.ms,namespace={}) count=int(mon.count[mn9]);spikes=count-previous_count;previous_count=count consumed=min(remaining,spikes);remaining-=consumed feedback.append({'startMs':window*P['feedback_ms'],'endMs':(window+1)*P['feedback_ms'],'gate':gate,'mn9Spikes':spikes,'consumed':consumed,'before':resource_before,'remaining':remaining}) actual_steps=np.rint(np.asarray(input_mon.t/b.ms)/P['dt_ms']).astype(np.int32) assert np.array_equal(actual_steps,steps)and np.array_equal(np.asarray(input_mon.i),channels) spike_indices=np.asarray(mon.i).astype(np.int32);spike_steps=np.rint(np.asarray(mon.t/b.ms)/P['dt_ms']).astype(np.int32) events=pd.DataFrame({'neuronIndex':spike_indices,'flywireId':ids[spike_indices],'step':spike_steps});events.to_parquet(OUT/f'{seed}-{condition}-spikes.parquet',index=False) assert sum(r['mn9Spikes']for r in feedback)==int((spike_indices==mn9).sum()) row={'seed':seed,'condition':condition,'remaining':remaining,'consumed':P['resource_units']-remaining,'mn9Spikes':previous_count,'totalSpikes':len(events),'incomingMn9Rows':len(incoming),'weightsSha256':weights_hash,'elapsedSeconds':time.monotonic()-start,'spikesSha256':sha(OUT/f'{seed}-{condition}-spikes.parquet'),'feedback':feedback} (OUT/f'{seed}-{condition}.json').write_text(json.dumps(row,indent=2)+'\n');trials.append(row) print(json.dumps({k:row[k]for k in ['seed','condition','consumed','mn9Spikes','totalSpikes','elapsedSeconds']}),flush=True) del net,neu,syn,mon,driver,inject,input_mon,original,expected,events;gc.collect() restores=[] for seed in P['seeds']: a=next(t for t in trials if t['seed']==seed and t['condition']=='intact');z=next(t for t in trials if t['seed']==seed and t['condition']=='mn9-restored') ea=pd.read_parquet(OUT/f'{seed}-intact-spikes.parquet');ez=pd.read_parquet(OUT/f'{seed}-mn9-restored-spikes.parquet') restores.append({'seed':seed,'allSpikeEventsEqual':ea.equals(ez),'feedbackEqual':a['feedback']==z['feedback'],'weightsEqual':a['weightsSha256']==z['weightsSha256']}) after=original_blobs() pd.DataFrame({'neuronIndex':np.arange(len(ids),dtype=np.int32),'flywireId':ids}).to_parquet(OUT/'neuron-index.parquet',index=False) summary={'executionStatus':'completed','startedAt':started,'finishedAt':datetime.now(timezone.utc).isoformat(),'protocol':P,'platform':platform.platform(),'python':sys.version.split()[0],'packages':{k:importlib.metadata.version(k)for k in ['brian2','numpy','pandas','pyarrow','Cython']},'inputIds':inputs,'before':before,'after':after,'trials':trials,'restoration':restores,'elapsedSeconds':time.monotonic()-began,'sourceSha256':sha(Path(__file__)),'protocolSha256':sha(ROOT/'protocol.json'),'maxRssKiB':resource.getrusage(resource.RUSAGE_SELF).ru_maxrss,'neuronIndexSha256':sha(OUT/'neuron-index.parquet')} (OUT/'summary.json').write_text(json.dumps(summary)+'\n') assert all(all(r[k]for k in ['allSpikeEventsEqual','feedbackEqual','weightsEqual'])for r in restores) print(json.dumps({'complete':True,'restoration':restores,'elapsedSeconds':summary['elapsedSeconds']}),flush=True)