| """Local CPU read-only implementation/gradient audit; no optimizer, no test EEG.""" |
| import os,sys,json,hashlib,datetime,ast |
| from pathlib import Path |
| os.environ.setdefault('MPLCONFIGDIR',str(Path('.cache/matplotlib').resolve()));os.environ.setdefault('HF_HOME',str(Path('.cache/huggingface').resolve())) |
| import numpy as np,torch |
| from bci_data import cached,preprocess,montage,LABELS |
| from foundation_models import FoundationClassifier,cs_topology |
| from reference_gpu import ReferenceModel,transform |
| from safetensors import safe_open |
|
|
| torch.set_num_threads(2);x,y,names=cached(1,'train');report={'utc':datetime.datetime.now(datetime.timezone.utc).isoformat(),'computation':'CPU on remote Lambda; no optimizer updates; no final-test EEG loaded','class_order':LABELS,'preprocessing':{},'trainability':[]} |
| for n in [64,32,16,8]: |
| idx=montage(names,n);a=preprocess(x[:5],idx);b=x[:5].copy();b[:,:,:128]+=9999;b[:,:,640:]-=9999;assert np.array_equal(preprocess(b,idx),a) |
| report['preprocessing'][str(n)]={'shape':list(a.shape),'sfreq':200,'indices':idx.tolist(),'names':[names[i] for i in idx],'outside_imagination_invariance':True,'median_channel_std_microvolt':float(np.median(a.std(-1))),'selected_CAR_max_abs_channel_mean':float(abs(a.mean(1)).max())} |
| |
| for kind in ['reve','csbrain']: |
| for mode in ['probe','partial','full']: |
| idx=montage(names,8);m=FoundationClassifier(kind,[names[i] for i in idx],mode);trainable=[name for name,p in m.named_parameters() if p.requires_grad] |
| backbone=[name for name in trainable if name.startswith('encoder.')];n_layers=22 if kind=='reve' else 12 |
| if mode=='probe':assert not backbone |
| if mode=='partial':assert all(any(f'layers.{k}.' in name for k in [n_layers-2,n_layers-1]) for name in backbone) |
| if mode=='full':assert all(p.requires_grad or name.endswith('mask_encoding') for name,p in m.named_parameters()) |
| report['trainability'].append({'model':kind,'mode':mode,'trainable_parameters':sum(p.numel() for p in m.parameters() if p.requires_grad),'total_parameters':sum(p.numel() for p in m.parameters()),'intended_mask_verified':True,'trainable_backbone_tensors':len(backbone)});del m |
| |
| regions,order=cs_topology(names);report['csbrain_topology']={'input_names':names,'regions':regions,'sorted_input_indices':order,'reordered_names':[names[i] for i in order],'all_64_input_channels_preserved':sorted(order)==list(range(64))} |
| |
| p=Path('checkpoints/reference3s_reve_full_pooled_64_seed-2026.pt');state=torch.load(p,map_location='cpu',weights_only=False);m=ReferenceModel('reve',names);m.load_state_dict(state['model']);m.eval();a=torch.tensor(transform(x[:5],np.arange(64),'reve'));logits=m(a);loss=torch.nn.functional.cross_entropy(logits,torch.tensor(y[:5]));loss.backward() |
| gradients={} |
| for prefix in ['encoder','query','norm_weight','head']: |
| g=[p.grad.detach().float() for name,p in m.named_parameters() if name.startswith(prefix) and p.grad is not None];gradients[prefix]={'l2_norm':float(torch.sqrt(sum((v*v).sum() for v in g))) if g else None,'tensors_with_gradient':len(g),'all_finite':all(torch.isfinite(v).all().item() for v in g)} |
| with safe_open('models/reve-base/model.safetensors',framework='pt',device='cpu') as weights:original_query=weights.get_tensor('cls_query_token') |
| report['reve_query']={'official_pretrained_query_l2':float(original_query.norm()),'unused_backbone_query_equal_to_pretrained':bool(torch.equal(state['model']['encoder.backbone.cls_query_token'],original_query)),'used_classifier_query_l2':float(state['model']['query'].norm()),'used_query_equal_to_pretrained':bool(torch.equal(state['model']['query'],original_query)),'finding':'SETU reference head uses separately random-initialized query. Official src/dt.py copies pretrained query into classifier. This is a reproducibility mismatch; causal effect on accuracy unmeasured.'} |
| report['reve_checkpoint_gradient_diagnostic']={'checkpoint_epoch':state['epoch'],'checkpoint_sha256':hashlib.sha256(p.read_bytes()).hexdigest(),'split':'training','subject':1,'trial_indices':list(range(5)),'batch_loss':float(loss.detach()),'gradients':gradients,'optimizer_steps':0,'historical_training_gradient_norms_available':False,'note':'Current-checkpoint eval-mode backward diagnostic; not historical BF16 training telemetry.'} |
| Path('reports/audit/accuracy_checks.json').write_text(json.dumps(report,indent=2));print('Accuracy audit checks saved',report['reve_query'],flush=True) |
|
|
