Buckets:
| """Export static FP32 ONNX and retain it only after CPU prediction parity.""" | |
| import argparse | |
| import hashlib | |
| import json | |
| from pathlib import Path | |
| import time | |
| from unittest.mock import patch | |
| import numpy as np | |
| from shapely.geometry import Polygon | |
| from training.model_factory import ROOT, configure_local_settings | |
| from inference import Model | |
| from load_dataset import dataset, load_sample | |
| from evaluation.quick_eval import semantic_array | |
| def export_model(component,checkpoint,destination): | |
| configure_local_settings() | |
| from ultralytics import YOLO | |
| import torch | |
| torch.set_num_threads(4) | |
| if destination.exists():raise FileExistsError(destination) | |
| wrapper=YOLO(str(checkpoint),task='obb' if component=='detection' else 'semantic') | |
| generated=Path(wrapper.export(format='onnx',imgsz=1024,batch=1,device='cpu',half=False, | |
| dynamic=False,simplify=False,opset=17,nms=False)) | |
| generated.replace(destination) | |
| return destination | |
| def compare_detection(left,right): | |
| if len(left.obb.cls)!=len(right.obb.cls):return False | |
| unused=set(range(len(right.obb.cls))) | |
| right_cls=right.obb.cls.cpu().numpy();right_scores=right.obb.conf.cpu().numpy() | |
| right_polygons=[Polygon(p) for p in right.obb.xyxyxyxy.cpu().numpy()] | |
| for points,category,score in zip(left.obb.xyxyxyxy.cpu().numpy(),left.obb.cls.cpu().numpy(),left.obb.conf.cpu().numpy()): | |
| polygon=Polygon(points) | |
| candidates=[] | |
| for index in unused: | |
| if right_cls[index]!=category:continue | |
| other=right_polygons[index];intersection=polygon.intersection(other).area | |
| union=polygon.area+other.area-intersection | |
| candidates.append((intersection/union if union>0 else 0,index)) | |
| if not candidates:return False | |
| iou,index=max(candidates) | |
| if iou<.999 or abs(float(score)-float(right_scores[index]))>.001:return False | |
| unused.remove(index) | |
| return True | |
| def main(): | |
| parser=argparse.ArgumentParser() | |
| parser.add_argument('--model-dir',type=Path,default=ROOT/'model') | |
| args=parser.parse_args() | |
| package=Model(args.model_dir) | |
| start=time.monotonic() | |
| manifest=json.loads((ROOT/'evaluation/quick_eval_manifest.json').read_text()) | |
| report={'runtime':'CPUExecutionProvider','precision':'FP32','input_size':1024,'components':{}} | |
| configure_local_settings() | |
| from ultralytics import YOLO | |
| import torch | |
| for component in ('detection','segmentation'): | |
| record=package.config['components'][component] | |
| checkpoint=package.directory/record['path'] | |
| destination=package.directory/('detector.onnx' if component=='detection' else 'segmenter.onnx') | |
| if destination.exists():raise FileExistsError(destination) | |
| if checkpoint.with_suffix('.onnx').exists():raise FileExistsError(checkpoint.with_suffix('.onnx')) | |
| result={'status':'FAILED','sample_count':0} | |
| try: | |
| export_model(component,checkpoint,destination) | |
| import onnxruntime as ort | |
| options=ort.SessionOptions();options.intra_op_num_threads=4 | |
| session=ort.InferenceSession(str(destination),sess_options=options,providers=['CPUExecutionProvider']) | |
| assert session.get_providers()==['CPUExecutionProvider'] | |
| onnx=YOLO(str(destination),task='obb' if component=='detection' else 'semantic') | |
| data=dataset(ROOT/'datasets/corpus',component,'val',license_scope='all') | |
| rows={r['id']:r for r in data.samples} | |
| disagreements=0;pixels=0;failures=[] | |
| for row in manifest['quick_eval'][component]: | |
| sample=load_sample(data.root,rows[row['sample_id']],data.expected_hashes) | |
| image=sample['image'][:,:,::-1].copy() | |
| kwargs={'conf':.001,'max_det':300} if component=='detection' else {} | |
| with torch.inference_mode(): | |
| left=package.models[component].predict(image,imgsz=1024,device='cpu',save=False,verbose=False,**kwargs)[0] | |
| # Reuse the explicitly bounded CPU session when the framework loads its backend. | |
| with patch('onnxruntime.InferenceSession', return_value=session): | |
| right=onnx.predict(image,imgsz=1024,device='cpu',save=False,verbose=False,**kwargs)[0] | |
| assert onnx.predictor.model.session.get_providers()==['CPUExecutionProvider'] | |
| if component=='detection': | |
| if not compare_detection(left,right):failures.append(row['sample_id']) | |
| else: | |
| a,b=semantic_array(left),semantic_array(right) | |
| if a.shape!=b.shape:raise ValueError('semantic export shape mismatch') | |
| disagreements+=int(np.count_nonzero(a!=b));pixels+=a.size | |
| result['sample_count']+=1 | |
| result.update(pixel_agreement=1-disagreements/pixels if pixels else None,failed_samples=failures) | |
| if failures or (pixels and disagreements/pixels>1e-4):raise ValueError('prediction parity tolerance exceeded') | |
| result.update(status='PASSED',sha256=hashlib.sha256(destination.read_bytes()).hexdigest(),bytes=destination.stat().st_size, | |
| tolerance='Detection: same class/count, matched polygon IoU>=.999, confidence delta<=.001; semantic pixel agreement>=.9999.') | |
| except Exception as error: | |
| result['error']=f'{type(error).__name__}: {error}' | |
| destination.unlink(missing_ok=True) | |
| checkpoint.with_suffix('.onnx').unlink(missing_ok=True) | |
| report['components'][component]=result | |
| report['elapsed_seconds']=time.monotonic()-start | |
| path=package.directory/'EXPORT.json' | |
| with path.open('x') as stream:json.dump(report,stream,indent=2) | |
| print(json.dumps(report)) | |
| if __name__=='__main__':main() | |
Xet Storage Details
- Size:
- 5.82 kB
- Xet hash:
- aae9f8e161c91f3efd9cfe1e417dce21496f1a9bf42555d42af2e2cc3541a474
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.