"""Four independent recorded-video trials; no fight mapping or Event calls.""" import copy import csv import functools import itertools import json import pathlib import subprocess import sys import time import traceback import cv2 import numpy as np import torch from PIL import Image ROOT=pathlib.Path('/eval') PREP=pathlib.Path('/work') NAME=sys.argv[1] rows=json.loads((ROOT/'inputs/pose.json').read_text()) torch.set_num_threads(2) torch.manual_seed(7) np.random.seed(7) torch.cuda.set_per_process_memory_fraction(.25) device=torch.device('cuda:0') start=time.monotonic() report=dict(candidate=NAME,status='STARTED',observation_fps=10,prediction_interval_seconds=.1, evaluation_samples=list(range(100,301)),source_display_frames=[242,723], ground_truth='User visual observation: upper-left two fight throughout both loops; others bystanders. Not independent clips.', camera_id=24,arbitrary_fight_mapping=False,metrics_are_offline=True) classes={int(r['output_index']):r['class_name'] for r in csv.DictReader(open(PREP/'reports'/('classes_'+NAME+'.csv')))} results=[] prior={} outpath=ROOT/'outputs'/(NAME+'.json') oldsummary=ROOT/'reports'/(NAME+'_summary.json') if outpath.exists() and oldsummary.exists(): old=json.loads(oldsummary.read_text()) if old['status']=='PASS': prior={r['sample']:r for r in json.loads(outpath.read_text())} archive=ROOT/'reports'/(NAME+'_initial_2fps.json') assert not archive.exists(), 'Do not repeat completed evaluation' archive.write_text(json.dumps(dict(summary=old,results=list(prior.values())))) report['reused_predictions']=len(prior) report['prior_wall_seconds']=old['wall_seconds'] def top5(scores,indices): order=np.argsort(-np.asarray(scores))[:5] return [dict(index=int(indices[k]),name=classes[int(indices[k])],score=float(scores[k])) for k in order] @functools.lru_cache(maxsize=80) def rgb(index): image=cv2.imread(str(ROOT/'inputs/frames'/f'{index:04d}.png')) assert image is not None return cv2.cvtColor(image,cv2.COLOR_BGR2RGB) def window(indices): assert min(indices)>=0 and max(indices)=.3)&np.isfinite(kp).all(axis=1) keypoints[slot,t,valid]=kp[valid,:2] scores[slot,t,valid]=kp[valid,2] seen[slot,t]=True return dict(keypoint=keypoints,keypoint_score=scores,img_shape=(1080,1920), total_frames=len(indices),start_index=0,label=-1),seen def pair_candidates(persons): out=[] for a,b in itertools.combinations(persons,2): if a['track_id'] is None or b['track_id'] is None or a['track_id']==b['track_id']: continue ba,bb=np.asarray(a['bbox']),np.asarray(b['bbox']) h=((ba[3]-ba[1])+(bb[3]-bb[1]))/2 ratio=np.linalg.norm((ba[:2]+ba[2:]-bb[:2]-bb[2:])/2)/max(h,1) if ratio<=1.8: out.append((sorted([a['track_id'],b['track_id']]),float(ratio))) return sorted(out) try: checkpoint=torch.load(PREP/'weights'/(NAME+'.pth'),map_location='cpu',weights_only=False) if NAME in ('stgcnpp','posec3d'): import mmaction.models import mmaction.datasets.transforms from mmaction.registry import MODELS from mmengine.config import Config from mmengine.dataset import Compose from mmengine.registry import init_default_scope init_default_scope('mmaction') config={'stgcnpp':'skeleton/stgcnpp/stgcnpp_8xb16-joint-u100-80e_ntu60-xsub-keypoint-2d.py', 'posec3d':'skeleton/posec3d/slowonly_r50_8xb16-u48-240e_ntu60-xsub-keypoint.py'}[NAME] cfg=Config.fromfile('/opt/mmaction2/configs/'+config) model=MODELS.build(cfg.model).eval().to(device) model.load_state_dict(checkpoint['state_dict'],strict=True) transform=Compose(cfg.val_pipeline) report.update(pipeline='official val_pipeline, one temporal view, no flip/TTA', missing='zero score/coordinates before official transforms; no interpolation, no ID stitching', joint_threshold=.3,score_activation='softmax over NTU60', temporal_samples=100 if NAME=='stgcnpp' else 48) if NAME=='stgcnpp': report['pair_rule']='bbox center distance / mean bbox height <= 1.8; spatial candidates only, not fight logic' for i in report['evaluation_samples']: if i in prior: results.append(prior[i]); continue indices=list(range(i-50,i+50)) if NAME=='stgcnpp' else list(range(i-24,i+24)) record=dict(sample=i,source_frame=rows[i]['source_frame'],seconds=rows[i]['seconds'], window=window(indices),items=[],input_people=len(rows[i]['persons']), no_track_candidates=sum(p['track_id'] is None for p in rows[i]['persons'])) candidates=pair_candidates(rows[i]['persons']) if NAME=='stgcnpp' else [(None,None)] for ids,ratio in candidates: data,seen=skeleton(indices,ids) if not data['keypoint_score'].any(): record['items'].append(dict(ids=ids,status='NO_VALID_SKELETON')) continue transformed=transform(copy.deepcopy(data)) inputs=transformed['inputs'].unsqueeze(0).to(device) with torch.inference_mode(): logits=model(inputs,mode='tensor',stage='head') prob=logits.softmax(-1).mean(0).cpu().numpy() assert prob.shape==(60,) and np.isfinite(prob).all() and abs(prob.sum()-1)<1e-5 coverage=seen.mean(axis=1).tolist() if ids is not None else None record['items'].append(dict(ids=ids,proximity_ratio=ratio,status='PARTIAL_TRACK' if coverage and min(coverage)<1 else 'OUTPUT', coverage=coverage,input_shape=list(inputs.shape),scores=prob.tolist(), valid_joint_fraction=float((data['keypoint_score']>0).mean()),top5=top5(prob,list(range(60))))) results.append(record) if (i-100)%50==0: print(NAME,i,len(record['items']),flush=True) elif NAME=='ava_slowonly': sys.path.insert(0,'/work') from ava_adapter import AvaAdapter,check_roi_contract check_roi_contract(device) model=AvaAdapter(checkpoint).eval().to(device) report.update(pipeline='official AVA4x16 offsets [-32,-16,0,16] on common 10FPS; shortest-side256, no crop', norm_mean=[123.675,116.28,103.53],norm_std=[58.395,57.12,57.375],rgb=True, external_bbox_filter='scenario person>=0.15 only; no AVA proposal0.9 filter', score_activation='sigmoid, exclude background0',mmcv_numerical_equivalence='NOT_TESTED', adapter='Official backbone/head with torchvision RoIAlign',roi_coordinate_check='PASS') w,h=455,256 for i in report['evaluation_samples']: if i in prior: results.append(prior[i]); continue indices=[i-32,i-16,i,i+16] imgs=np.stack([cv2.resize(rgb(j),(w,h),interpolation=cv2.INTER_LINEAR) for j in indices]).astype(np.float32) imgs=(imgs-np.array(report['norm_mean'],np.float32))/np.array(report['norm_std'],np.float32) x=torch.from_numpy(imgs.transpose(3,0,1,2).copy()[None]).to(device) persons=rows[i]['persons'] boxes=np.array([p['bbox'] for p in persons],np.float32).reshape(-1,4) boxes[:,0::2]=np.clip(boxes[:,0::2],0,1920)*w/1920 boxes[:,1::2]=np.clip(boxes[:,1::2],0,1080)*h/1080 assert np.all(boxes[:,2:]>=boxes[:,:2]) rois=torch.from_numpy(np.column_stack([np.zeros(len(boxes)),boxes]).astype(np.float32)).to(device) record=dict(sample=i,source_frame=rows[i]['source_frame'],seconds=rows[i]['seconds'],window=window(indices),items=[]) if len(boxes): with torch.inference_mode(): scores=model(x,rois).sigmoid().cpu().numpy()[:,1:] assert scores.shape==(len(persons),80) and np.isfinite(scores).all() for p,s in zip(persons,scores): record['items'].append(dict(track_id=p['track_id'],query=p['query'],bbox=p['bbox'], detection_score=p['score'],status='OUTPUT',scores=s.tolist(),top5=top5(s,list(range(1,81))))) results.append(record) if (i-100)%50==0: print(NAME,i,len(record['items']),flush=True) elif NAME=='yowov2': sys.path.insert(0,'/opt/yowov2') from config.yowo_v2_config import yowo_v2_config from models.yowo.yowo import YOWO from dataset.transforms import BaseTransform transform=BaseTransform(img_size=224) model=YOWO(yowo_v2_config['yowo_v2_large'],device,num_classes=80,conf_thresh=.1,nms_thresh=.5,topk=40,trainable=False,multi_hot=True) model.load_state_dict(checkpoint['model'],strict=True) model.eval().to(device) report.update(pipeline='official BaseTransform RGB/PIL resize224, tensors0..255; consecutive16, last frame is keyframe', objectness_threshold=.1,nms_iou=.5,pre_nms_topk=40,score_activation='separate action sigmoid; no objectness multiplication', identities='per-output index Y1..Yn, not tracking identity') for i in report['evaluation_samples']: if i in prior: results.append(prior[i]); continue indices=list(range(i-15,i+1)) clip,_=transform([Image.fromarray(rgb(j)) for j in indices]) x=torch.stack(clip,dim=1).unsqueeze(0).to(device) assert x.shape==(1,3,16,224,224) and float(x.max())>1 with torch.inference_mode(): out=model(x)[0] assert out.ndim==2 and out.shape[1]==85 and np.isfinite(out).all() record=dict(sample=i,source_frame=rows[i]['source_frame'],seconds=rows[i]['seconds'],window=window(indices),items=[]) for j,o in enumerate(out): assert np.max(o[:4])<=1.01 and np.min(o[:4])>=-.01 bbox=(o[:4]*[1920,1080,1920,1080]).tolist() s=o[5:] record['items'].append(dict(display_id='Y'+str(j+1),bbox=bbox,objectness=float(o[4]),status='OUTPUT', scores=s.tolist(),top5=top5(s,list(range(80))))) results.append(record) if (i-100)%50==0: print(NAME,i,len(record['items']),flush=True) else: raise ValueError(NAME) torch.cuda.synchronize() assert len(results)==201 for r in results: r['window']=window(r['window']['samples']) for r in results: for item in r['items']: if 'scores' in item: assert all(0<=v<=1 for v in item['scores']) and len(item['top5'])==5 report.update(status='PASS',prediction_count=len(results),output_items=sum(len(r['items']) for r in results), max_items=max(len(r['items']) for r in results),partial_tracks=sum(it['status']=='PARTIAL_TRACK' for r in results for it in r['items'])) except Exception as e: report.update(status='BLOCKED',error=repr(e),traceback=traceback.format_exc()) finally: report.update(wall_seconds=round(time.monotonic()-start,3),peak_cuda_allocated_bytes=torch.cuda.max_memory_allocated(), peak_cuda_reserved_bytes=torch.cuda.max_memory_reserved()) (ROOT/'outputs'/(NAME+'.json')).write_text(json.dumps(results)) (ROOT/'reports'/(NAME+'_summary.json')).write_text(json.dumps(report,indent=2)) print(json.dumps(report),flush=True) raise SystemExit(0 if report['status']=='PASS' else 1)