"""INFERFPS-01: run-local PTS selection and bounded, non-waiting measurements.""" from collections import Counter, OrderedDict, deque import math import threading import time def analysis_fps(value): if type(value) not in (int, float) or not math.isfinite(value) or not 1 <= value <= 60: raise ValueError('Analysis FPS must be finite and between 1 and 60.') return float(value) class PtsSampler: def __init__(self, fps=10): self.fps = analysis_fps(fps) self.origin = self.last = self.slot = None def select(self, pts): if type(pts) is not int or not 0 <= pts < (1 << 64) - 1: return False, 'invalid_pts' if self.last is not None and pts == self.last: return False, 'duplicate_pts' reset = self.last is not None and pts < self.last if self.origin is None or reset: self.origin, self.slot = pts, -1 self.last = pts slot = math.floor((pts - self.origin) * self.fps / 1_000_000_000) if slot <= self.slot: return False, 'sampling_drop' self.slot = slot return True, 'pts_reset' if reset else 'selected' class AnalysisMetrics: STAGES = frozenset(('decoded', 'selected', 'sampling_drop', 'invalid_pts', 'duplicate_pts', 'pts_reset', 'preprocess', 'infer', 'native', 'result', 'publisher_submit')) PAIRS = {'preprocess': 'selected', 'infer': 'preprocess', 'native': 'infer', 'result': 'native', 'publisher_submit': 'result'} DURATIONS = frozenset(('queue', 'tracking', 'tracking_cpu', 'scenario', 'scenario_cpu', 'result_work', 'result_work_cpu', 'collector_wait', 'collector_work', 'collector_work_cpu', 'collector_evict', 'collector_evict_cpu', 'result_publish', 'result_publish_cpu', 'result_lifecycle', 'result_lifecycle_cpu')) def __init__(self, fps=10, clock=time.monotonic_ns): self.target = analysis_fps(fps) self.clock = clock self._lock = threading.Lock() self._counts = Counter() self._first, self._last = {}, {} self._last_key = {} self._frames = OrderedDict() self._latencies = {} self._until = 0 self._closed = False self.missed = 0 self._tracking_profiles = [] self._tracking_reserved = 0 self._window_revision = 0 def configure(self, enabled, seconds=60): if type(enabled) is not bool or type(seconds) is not int or not 1 <= seconds <= 60: raise ValueError('Invalid bounded measurement window.') with self._lock: if self._closed: raise RuntimeError('Measurement run is closed.') self._until = self.clock() + seconds * 1_000_000_000 if enabled else 0 self._frames.clear() self._latencies.clear() self._tracking_profiles.clear() self._tracking_reserved = 0 self._window_revision += 1 def reserve_tracking_profile(self): if self._closed or self.clock() >= self._until or self._tracking_reserved >= 3: return None with self._lock: if self._closed or self.clock() >= self._until or self._tracking_reserved >= 3: return None self._tracking_reserved += 1 return self._window_revision def detail_enabled(self): return not self._closed and self.clock() < self._until def record_tracking_profile(self, revision, value): with self._lock: if not self._closed and revision == self._window_revision and self.clock() < self._until: if len(self._tracking_profiles) < 3: self._tracking_profiles.append(value) def mark(self, stage, key=None): if stage not in self.STAGES: return if not self._lock.acquire(blocking=False): self.missed += 1 return try: if self._closed: return now = self.clock() self._counts[stage] += 1 self._first.setdefault(stage, now) self._last[stage] = now if isinstance(key, tuple) and len(key) == 2 and all(type(v) is int for v in key): self._last_key[stage] = key if now >= self._until: self._frames.clear() return if key is None: return if stage == 'selected': if len(self._frames) >= 256: self._frames.popitem(last=False) self._frames[key] = {'selected': now} frame = self._frames.get(key) if frame is None: return previous = self.PAIRS.get(stage) if previous in frame: self._latencies.setdefault(previous + '_to_' + stage, deque(maxlen=256)).append( now - frame[previous]) frame[stage] = now if stage == 'publisher_submit': self._frames.pop(key, None) finally: self._lock.release() def duration(self, stage, elapsed_ns): if stage not in self.DURATIONS or elapsed_ns < 0: return if not self._lock.acquire(blocking=False): self.missed += 1 return try: if not self._closed and self.clock() < self._until: self._latencies.setdefault(stage, deque(maxlen=256)).append(elapsed_ns) finally: self._lock.release() def snapshot(self): with self._lock: now = self.clock() if now >= self._until: self._frames.clear() counts, first, last = dict(self._counts), dict(self._first), dict(self._last) keys = {k: list(v) for k, v in self._last_key.items()} values = {k: list(v) for k, v in self._latencies.items()} detail = not self._closed and now < self._until pending = len(self._frames) tracking_profiles = list(self._tracking_profiles) durations = {} for stage, samples in values.items(): samples.sort() durations[stage] = dict(samples=len(samples), mean_ms=sum(samples)/len(samples)/1e6, p95_ms=samples[math.ceil(len(samples)*.95)-1]/1e6, max_ms=samples[-1]/1e6) return dict(analysisFpsTarget=self.target, sampled_ns=now, counts=counts, first_ns=first, last_ns=last, detail=detail, pending_samples=pending, last_frame=keys, observation_missed=self.missed, durations=durations, sourceFpsObserved=None, closed=self._closed, tracking_profiles=tracking_profiles) def close(self): with self._lock: self._closed = True self._until = 0 self._frames.clear() self._tracking_profiles.clear()