File size: 4,679 Bytes
98bde72 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 | """Command-line entry points for specifications, episodes, and analysis."""
import argparse,json,sys
from pathlib import Path
import numpy as np
from .schema import DesignSpec,canonical
def main():
p=argparse.ArgumentParser(prog='peppa');sub=p.add_subparsers(dest='command',required=True)
s=sub.add_parser('validate');s.add_argument('spec')
s=sub.add_parser('run');s.add_argument('--spec',required=True);s.add_argument('--registry',required=True);s.add_argument('--controller',required=True);s.add_argument('--trace',required=True)
s=sub.add_parser('replay');s.add_argument('trace');s.add_argument('--output')
s=sub.add_parser('train-ptm');s.add_argument('data');s.add_argument('--output',required=True);s.add_argument('--l2',type=float,default=.001)
s=sub.add_parser('fit-ternary');s.add_argument('data');s.add_argument('--output',required=True)
s=sub.add_parser('normalize-snooppi');s.add_argument('data');s.add_argument('--output',required=True)
s=sub.add_parser('split');s.add_argument('data');s.add_argument('--output',required=True)
s=sub.add_parser('summarize');s.add_argument('data');s.add_argument('--value',default='joint_success');s.add_argument('--group',default='task');s.add_argument('--output',required=True)
s=sub.add_parser('plan');s.add_argument('--output',required=True)
args=p.parse_args()
def read(path):return json.loads(Path(path).read_text())
def write(path,x):
Path(path).parent.mkdir(parents=True,exist_ok=True);Path(path).write_text(json.dumps(x,indent=2,allow_nan=False)+'\n')
if args.command=='validate':
spec=DesignSpec.model_validate(read(args.spec));print(canonical(spec));return
if args.command=='replay':
from .trace import replay
state=replay(args.trace)
if args.output:write(args.output,state)
else:print(json.dumps(state,indent=2))
elif args.command=='run':
from .engine import Engine
from .builtin import registry_from_config
from .controllers import APIController,ScriptedController,AnthropicController
spec=DesignSpec.model_validate(read(args.spec));cfg=read(args.controller)
controller=ScriptedController(cfg['decisions']) if cfg['type']=='scripted' else AnthropicController(**cfg['settings']) if cfg['type']=='anthropic' else APIController(**cfg['settings'])
state=Engine(spec,registry_from_config(read(args.registry)),args.trace).run(controller)
print(json.dumps({'stopped':state['stopped'],'candidates':len(state['candidates']),'spent':state['spent'],'errors':len(state['errors'])}))
elif args.command=='train-ptm':
from .ptm import fit
with np.load(args.data,allow_pickle=False) as d:
if not np.all(d['split']=='train'):raise ValueError('training file contains nontraining rows')
model,report=fit(d['binder'],d['target'],d['labels'],d['pairs'],l2=args.l2)
Path(args.output).parent.mkdir(parents=True,exist_ok=True);model.save(args.output);write(args.output+'.json',report)
elif args.command=='fit-ternary':
from .ternary import fit_cooperativity
d=read(args.data);write(args.output,fit_cooperativity(np.array(d['totals_nm']),d['observed_abl_nm'],d['kd_a_nm'],d['kd_b_nm'],d.get('sd_nm')))
elif args.command=='normalize-snooppi':
from .data import normalize_snooppi
rows=[json.loads(x) for x in Path(args.data).read_text().splitlines() if x]
Path(args.output).parent.mkdir(parents=True,exist_ok=True);Path(args.output).write_text(''.join(json.dumps(normalize_snooppi(x))+'\n' for x in rows))
elif args.command=='split':
from .data import connected_splits
rows=read(args.data);write(args.output,[dict(row,split=sp) for row,sp in zip(rows,connected_splits(rows))])
elif args.command=='summarize':
from .metrics import grouped_bootstrap
write(args.output,grouped_bootstrap(read(args.data),args.value,args.group))
elif args.command=='plan':
rows=[]
arms=['fixed', 'react_muse', 'compiled_reduced_checks', 'peppa_muse', 'peppa_astra', 'peppa_claude', 'peppa_no_retrieval']
for family in ['affinity_specificity','motif','conformation']:
for task in range(1,9):
for arm in arms:
for seed in [2027,2028,2029]:
rows.append({'task':f'{family}_{task:02d}','family':family,'arm':arm,'seed':seed,'status':'awaiting frozen task manifest','raw_proposals':384,'computational_cycles':2,'structures_per_cycle':18,'gpu_minutes':1440,'controller_tokens':128000,'ranking_batch':12})
write(args.output,rows)
if __name__=='__main__':main()
|