"""PIPECORE-01: decoded ingress가 identity를 발급하고 SDK FrameMeta가 운반한다.""" from dataclasses import asdict from functools import lru_cache from collections import deque import ctypes import json import threading import time from app.contracts.frames import FrameEnvelope, FrameTime from app.frame_time import FrameAnchor, RtpFrameClock, validate_carried_time @lru_cache(maxsize=1) def identity_meta_type(): register = ctypes.CDLL('libnvds_meta.so').nvds_get_user_meta_type register.argtypes = [ctypes.c_char_p] register.restype = ctypes.c_int return register(b'CONTEXTRA.FRAME.IDENTITY.V1') @lru_cache(maxsize=1) def rtp_meta_type(): register = ctypes.CDLL('libnvds_meta.so').nvds_get_user_meta_type register.argtypes = [ctypes.c_char_p] register.restype = ctypes.c_int return register(b'CONTEXTRA.RTP.FRAME.V1') class FrameIngress: def __init__(self, server_id, source_id, stream_id, run_id, generation, config_version=None, clock_ns=time.time_ns, frame_time_anchor=None, analysis_fps_target=10): self.identity = (server_id, source_id, stream_id, run_id, generation) if any(type(x) is not str or not x or len(x) > 256 for x in self.identity[:4]): raise ValueError('Frame identity must be explicitly configured.') if type(generation) is not int or generation < 1: raise ValueError('Invalid frame generation.') self.config_version = config_version from app.test_frame_time import synthetic_anchor self.frame_time = (FrameAnchor.model_validate(frame_time_anchor).frame_time() if frame_time_anchor is not None else synthetic_anchor(source_id)) self.rtp_clock = RtpFrameClock(self.frame_time) self.clock_ns = clock_ns self._seq = 0 self._lock = threading.Lock() self.track_result_gaps = False from app.input_timing import source_latest_enabled self.source_latest = source_latest_enabled() self._source_gap_pending = False self._selected_frames = deque(maxlen=512) self._selection_gap = False from app.analysis_metrics import PtsSampler, AnalysisMetrics self.sampler = PtsSampler(analysis_fps_target) self.metrics = AnalysisMetrics(analysis_fps_target) def stamp(self, batch, *, selected_monotonic_ns=None): decisions = [] with self._lock: for frame in batch.frame_items: if any(True for _ in frame.user_meta_items(identity_meta_type())): raise ValueError('Frame identity already assigned.') pts = int(frame.buffer_pts) packets = [m.get_user_data_json() for m in frame.user_meta_items(rtp_meta_type())] if len(packets) > 1: raise ValueError('Ambiguous Frame RTP metadata.') frame_time = self.rtp_clock.resolve(packets[0] if packets else None) if self.source_latest: from app.input_timing import gap_meta_type if any(True for _ in frame.user_meta_items(gap_meta_type())): self._source_gap_pending = True self.metrics.mark('source_queue_gap') sequence = self._seq self._seq += 1 key = (int(frame.frame_number), pts) self.metrics.mark('decoded') previous_slot = self.sampler.slot selected, reason = self.sampler.select(None if pts == (1 << 64) - 1 else pts) if selected and self._source_gap_pending: # Raw frames skipped within normal FPS sampling are not missing analysis samples. if previous_slot is not None and self.sampler.slot > previous_slot + 1: self._selection_gap = True self.metrics.mark('source_analysis_gap') self._source_gap_pending = False decisions.append(selected) if reason != 'selected': self.metrics.mark(reason) if self.track_result_gaps: self._record_selection(sequence, selected, reason) if not selected: continue # Sampling must not skip the RTP clock or renumber decoded Frames. envelope = FrameEnvelope( *self.identity, sequence, None if pts == (1 << 64) - 1 else pts, self.clock_ns() // 1_000_000, int(frame.source_width), int(frame.source_height), self.config_version, frame_time, ) value = dict(schema=1, frame=asdict(envelope), sdk_frame=int(frame.frame_number), sdk_pts=pts) if selected_monotonic_ns is not None: if type(selected_monotonic_ns) is not int or selected_monotonic_ns <= 0: raise ValueError('Invalid private Frame timing.') value['selected_monotonic_ns'] = selected_monotonic_ns append_identity(batch, frame, value) self.metrics.mark('selected', key) # Current canonical path owns a single-camera batch. Never discard a shared batch. if len(decisions) != 1: raise ValueError('PTS selection requires the canonical single-frame batch.') return decisions[0] def _record_selection(self, sequence, selected, reason): if reason in ('invalid_pts', 'duplicate_pts', 'pts_reset'): self._selection_gap = True if selected: if len(self._selected_frames) == self._selected_frames.maxlen: self._selection_gap = True self._selected_frames.append((sequence, self._selection_gap)) self._selection_gap = False def result_gap(self, frame): """FALL-WIRE-01: compare accepted selections, never decoded frame_seq jumps.""" with self._lock: gap = False while self._selected_frames and self._selected_frames[0][0] < frame.frame_seq: self._selected_frames.popleft() gap = True if not self._selected_frames or self._selected_frames[0][0] != frame.frame_seq: return True _, selected_gap = self._selected_frames.popleft() return gap or selected_gap def copy(self, frame, *, with_timing=False): values = [m.get_user_data_json() for m in frame.user_meta_items(identity_meta_type())] if len(values) != 1 or not isinstance(values[0], dict): raise ValueError('Frame identity missing or duplicated.') value = values[0] if (value.get('schema') != 1 or value.get('sdk_frame') != int(frame.frame_number) or value.get('sdk_pts') != int(frame.buffer_pts)): raise ValueError('Frame identity transport mismatch.') data = value.get('frame') if not isinstance(data, dict) or set(data) != set(FrameEnvelope.__dataclass_fields__): raise ValueError('Invalid Frame identity schema.') if (tuple(data.get(k) for k in ('server_id', 'source_id', 'stream_id', 'run_id', 'run_generation')) != self.identity or data['config_version'] != self.config_version): raise ValueError('Frame belongs to another run.') if (type(data['frame_seq']) is not int or data['frame_seq'] < 0 or type(data['ingest_time_wall_ms']) is not int or data['pts_ns'] != (None if value['sdk_pts'] == (1 << 64) - 1 else value['sdk_pts']) or type(data['frame_width']) is not int or type(data['frame_height']) is not int): raise ValueError('Invalid Frame identity values.') data = dict(data, frame_time=validate_carried_time(data['frame_time'], self.frame_time)) envelope = FrameEnvelope(**data) if (envelope.frame_width != int(frame.source_width) or envelope.frame_height != int(frame.source_height)): raise ValueError('Source dimensions changed across inference.') if with_timing: selected = value.get('selected_monotonic_ns') if type(selected) is not int or not 0 < selected <= time.monotonic_ns(): raise ValueError('Missing or invalid private Frame timing.') return envelope, selected return envelope def require_identity_binding(): try: import contextra_frame_meta as binding except ImportError as error: raise RuntimeError('Native frame identity binding is not packaged.') from error if getattr(binding, 'sdk_abi', None) != '__pybind11_internals_v5_gcc_libstdcpp_cxxabi1018__': raise RuntimeError('Native frame identity binding ABI is not validated.') return binding.append_json def append_identity(batch, frame, value): # SDK Python append는 generic UserMetadata를 지원하지 않는다. public C++ overload 사용. require_identity_binding()(batch, frame, json.dumps(value, ensure_ascii=True, allow_nan=False), identity_meta_type())