| """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() |
|
|
