import ctypes import importlib.util import json import os import pathlib import statistics import sys import time from types import SimpleNamespace import numpy as np import scipy.linalg import app.byte_tracking as ctx from app.contracts.frames import FrameEnvelope, DetectionObservation R=pathlib.Path('/evidence') spec=importlib.util.spec_from_file_location('ssims_tracking','/ssims/byte_tracker.py') ssims=importlib.util.module_from_spec(spec);sys.modules[spec.name]=ssims;spec.loader.exec_module(ssims) backend=[] for path in sorted({line.split()[-1] for line in pathlib.Path('/proc/self/maps').read_text().splitlines() if 'openblas' in line and '/' in line}): lib=ctypes.CDLL(path) for symbol in ('openblas_get_num_threads','openblas_get_num_threads64_','scipy_openblas_get_num_threads','scipy_openblas_get_num_threads64_'): try: fn=getattr(lib,symbol) except AttributeError: continue fn.restype=ctypes.c_int;backend.append(dict(library=pathlib.Path(path).name,threads=fn()));break assert backend and all(b['threads']==1 for b in backend) def fixtures(kind): rows=[] for seq in range(100): frame=FrameEnvelope('test','test','test','test',1,seq,seq*100000000,0,1920,1080) n,classes=(12,2) if kind=='few' else (300,12) if kind=='class_churn': n,classes=(80,80) if seq<10 else (0,80) if seq<60 else (4,2) elif seq in range(30,35): n=0 elif seq in range(65,70): n//=2 values=[] for i in range(n): score=.9 if kind=='class_churn' or i%5==0 else .2 if i%5==1 else .05 x=(i%20)*.04+seq*.00001;y=(i//20)*.04 values.append(DetectionObservation(frame,'RFDETRMedium',i%classes,score,(x,y,x+.025,y+.025))) rows.append(tuple(values)) return rows def summary(values): return {'mean_ms':statistics.mean(values), 'p95_ms':float(np.percentile(values,95)), 'n':len(values)} def contextra(frames): tracker=ctx.ByteTracker(ctx.TrackingSettings(frame_rate=10)) original=tracker.sv.ByteTrack times={k:[] for k in ('prepare','core','library_mapping','adapter_other','total')} current={};calls=[];states=[];outputs=[] class Wrapped: def __init__(self,**kw): self.obj=original(**kw) method=self.obj.update_with_tensors def core(*args,**kwargs): t=time.perf_counter_ns() try:return method(*args,**kwargs) finally:current['core']+=time.perf_counter_ns()-t;current['calls']+=1 self.obj.update_with_tensors=core def update_with_detections(self,d): t=time.perf_counter_ns();before=current['core'] try:return self.obj.update_with_detections(d) finally:current['mapping']+=time.perf_counter_ns()-t-(current['core']-before) def __getattr__(self,k):return getattr(self.obj,k) tracker.sv=SimpleNamespace(ByteTrack=Wrapped,Detections=tracker.sv.Detections) prepare=ctx._prepare_observations def measured_prepare(*args): t=time.perf_counter_ns() try:return prepare(*args) finally:current['prepare']+=time.perf_counter_ns()-t ctx._prepare_observations=measured_prepare try: for frame in frames: current.update(core=0,mapping=0,prepare=0,calls=0) t=time.perf_counter_ns();outputs.append(tracker.update(frame));total=time.perf_counter_ns()-t vals=dict(prepare=current['prepare'],core=current['core'],library_mapping=current['mapping'], adapter_other=total-current['prepare']-current['core']-current['mapping'],total=total) for k,v in vals.items():times[k].append(v/1e6) calls.append(current['calls']) states.append({'tracked':sum(len(t.tracked_tracks) for t in tracker._trackers.values()), 'lost':sum(len(t.lost_tracks) for t in tracker._trackers.values()),'classes':len(tracker._trackers)}) tracker.reset();current.update(core=0,mapping=0,prepare=0,calls=0) assert tracker.update(frames[0])==outputs[0] tracker.reset() finally:ctx._prepare_observations=prepare return {'timings':{k:summary(v[10:]) for k,v in times.items()},'tail_total':summary(times['total'][70:]), 'core_calls_mean':statistics.mean(calls[10:]),'states':states,'outputs':outputs} def ssims_run(frames): params={'track_high_thresh':.25,'track_low_thresh':.1,'new_track_thresh':.25, 'match_thresh':.8,'track_buffer':30} tracker=ssims.PoseByteTrack(mode='ultra',params_overrides=params,frame_rate=10) core=tracker.tracker.update current={};times={k:[] for k in ('prepare','core','output_mapping','total')};states=[] def measured(*args): current['core_start']=time.perf_counter_ns() try:return core(*args) finally:current['core_end']=time.perf_counter_ns() tracker.tracker.update=measured for frame in frames: t=time.perf_counter_ns() data=[{'bbox':(o.bbox_xyxy[0]*1920,o.bbox_xyxy[1]*1080,o.bbox_xyxy[2]*1920,o.bbox_xyxy[3]*1080), 'score':float(np.float32(o.score)),'cls':o.class_id} for o in frame] # Same float32 boxes as Contextra, preserving list input used by SSIMS caller. for d in data:d['bbox']=tuple(np.asarray(d['bbox'],dtype=np.float32)) result=tracker.update(data) end=time.perf_counter_ns() vals=dict(prepare=current['core_start']-t,core=current['core_end']-current['core_start'], output_mapping=end-current['core_end'],total=end-t) for k,v in vals.items():times[k].append(v/1e6) states.append({'tracked':len(tracker.tracked),'lost':len(tracker.tracker.lost),'outputs':len(result)}) return {'timings':{k:summary(v[10:]) for k,v in times.items()},'tail_total':summary(times['total'][70:]), 'states':states,'core_calls_mean':1,'params':params} out={'backend':backend,'frames':100,'warmup':10,'frame_rate':10,'results':{}} for kind in ('few','mixed','class_churn'): frames=fixtures(kind) out['results'][kind]={'contextra':contextra(frames),'ssims':ssims_run(frames), 'candidate_counts':[len(f) for f in frames]} (R/(sys.argv[1]+'.json')).write_text(json.dumps(out,indent=2)) print(json.dumps({k:{name:value['timings'] for name,value in v.items() if name!='candidate_counts'} for k,v in out['results'].items()},indent=2))