"""Native gameplay measurement of actual accepted-pass plans and catch releases.""" from pathlib import Path from concurrent.futures import ThreadPoolExecutor import argparse, hashlib, json, os, subprocess, math ap=argparse.ArgumentParser(description=__doc__) ap.add_argument('rom',type=Path) ap.add_argument('output',type=Path) ap.add_argument('--baseline',type=Path,required=True) ap.add_argument('--state',type=Path,required=True) ap.add_argument('--tools',type=Path,default=Path(__file__).resolve().parent) ap.add_argument('--minimum-gap',type=int,choices=[0,16,24],default=0) a=ap.parse_args();h=a.tools.resolve();o=a.output.resolve();o.mkdir(parents=True,exist_ok=True) source=(h/'ai_playbook_gameplay_hook.py').read_text() source=source.replace('put(own+5,0,-246);put(opp+5,0,246)',"put(own+5,0,-246);put(opp+5,0,246)") source=source.replace('px,depth=110,180;put(carrier,px,depth)',"px,depth=64*s.get('sign',1),220;put(carrier,px,depth)") source=source.replace("ww(0xf060,0);ww(0xf002,0)","ww(0xf060,0);ww(0xf002,0);ww(0xf08c,0);ww(0xf08e,0)") source=source.replace('put(own,0,220)',"put(own,-48*s.get('sign',1),240)") source+=''' base_tick=test_tick def test_tick(f): base_tick(f) if f>108: rows[-1]['pass_plan_pending']=rw(0xf08c) rows[-1]['pass_plan_ttl']=rw(0xf08e) rows[-1]['native_aim_index']=rw(0xbedc) if f==708:(p/'trajectory.json').write_text(json.dumps(rows,indent=2)+'\\n') ''' (o/'hook.py').write_text(source) specs=[dict(kind='finish',team=t,direction=d,sign=x,difficulty=1) for t in (0,1) for d in (-1,1) for x in (-1,1)] specs += [dict(kind='finish',team=0,direction=d,sign=1,difficulty=level) for d in (-1,1) for level in (0,2)] def run(profile,rom,index,s): p=o/(profile+'-'+str(index));p.mkdir(exist_ok=True);(p/'spec.json').write_text(json.dumps(s));(p/'commands.txt').write_text(f'100 load {a.state.resolve()}\n148 snap\n188 snap\n248 snap\n708 snap\n709 exit\n') env={k:v for k,v in os.environ.items() if not k.startswith('GPGX_')};env.update(GPGX_DEV='513',GPGX_HOOK=str(o/'hook.py')) with (p/'log.txt').open('w') as log:subprocess.run(['nice','-n','19','python3',str(h/'test_core.py'),str(rom.resolve()),str(p/'commands.txt'),str(p)],env=env,stdout=log,stderr=log,check=True,timeout=60) rr=json.loads((p/'trajectory.json').read_text());assert len(rr)==600 and 011: assert now['native_aim_index']<9,('non-native shot target',s,now['native_aim_index']) releases.append(now['frame']) return dict(plan_frames=[x['frame'] for x in pending],native_buffered_release_frames=releases,final_score=rr[-1]['score']) def pair(item): i,s=item;return dict(spec=s,baseline=run('old',a.baseline,i,s),candidate=run('new',a.rom,i,s)) with ThreadPoolExecutor(max_workers=3) as pool:rows=list(pool.map(pair,enumerate(specs))) summary=dict(scenarios=len(rows),hard_cases_with_new_plan=sum(bool(x['candidate']['plan_frames']) for x in rows if x['spec']['difficulty']==1),hard_cases_with_native_release=sum(bool(x['candidate']['native_buffered_release_frames']) for x in rows if x['spec']['difficulty']==1),normal_easy_excluded=4) (o/'results.json').write_text(json.dumps(dict(rom_sha256=hashlib.sha256(a.rom.read_bytes()).hexdigest(),baseline_sha256=hashlib.sha256(a.baseline.read_bytes()).hexdigest(),summary=summary,cases=rows),indent=2)+'\n') print(json.dumps(summary),flush=True) retained=[] for i,case in enumerate(rows[:8]): rr=json.loads((o/('new-'+str(i))/'trajectory.json').read_text()) for k,(before,now) in enumerate(zip(rr,rr[1:])): target=before['cpu_pending']-1 if 0<=target<12 and not now['cpu_pending'] and now['owner']==target: future=rr[k+2:k+122] moved=any(r['owner']==target and math.dist(r['positions'][target],now['positions'][target])>4 for r in future) passed=any(r['owner']>11 and r['passer']==target and r['target']<12 for r in future) shot=any(r['owner']>11 and r['shooter']==target and r['animations'][target]==0x7fc for r in future) # BED8/7FC describe a forehand buffered shot; ordinary native BC78/C3C2 # releases also stamp BEDA and the FF|actor owner token. Preserve all old # predicates and accept this additional positively observed native action. native_release=any(now['passer']!=target and r['passer']==target and r['owner']==(0xff00|target) and (r['target']<12 or 0xff00<=r['target']<0xff0c or r['target']==65535) and math.dist(r['puck'],now['puck'])>4 for r in future) retained.append(dict(case=i,frame=now['frame'],native_owned_skating=moved,native_pass=passed,native_shot=shot,native_unbuffered_release=native_release)) assert summary['hard_cases_with_native_release']>=1,'no actual Native buffered release observed' if a.minimum_gap: assert retained and any(x['native_owned_skating'] or x['native_pass'] or x['native_shot'] or x['native_unbuffered_release'] for x in retained),'no ordinary Native continuation after retaining a buffered catch' # Native input/target/physics equality is checked in the contract suites. # Host-frame timings can shift with CPU cycle cost; retain all live controls. controls=[] for i in range(8,12): old=json.loads((o/('old-'+str(i))/'trajectory.json').read_text());new=json.loads((o/('new-'+str(i))/'trajectory.json').read_text());first=next((k for k,(aa,bb) in enumerate(zip(old,new)) if aa!=bb),None) controls.append(dict(case=i,identical_live_trace=first is None,first_difference_frame=None if first is None else old[first]['frame'],first_difference_fields=[] if first is None else [k for k in old[first] if old[first][k]!=new[first][k]])) report=json.loads((o/'results.json').read_text());report.update(retained_native_continuations=retained,normal_easy_live_controls=controls);(o/'results.json').write_text(json.dumps(report,indent=2)+'\n') print('PASS actual Native buffered release/target range/clock/controller continuity and retained-catch skating/passing; timing controls retained')