"""PARSER-CUTOVER-01: SDK iterator 수명 안에서 숫자만 복사한다. Native rect는 mux/surface pixel LTWH, canonical bbox는 normalized XYXY다. 기존 tracker/storage의 CXCYWH는 전달 시에만 변환한다. Pixel은 보관하지 않는다. """ from dataclasses import dataclass from math import isfinite from functools import lru_cache @lru_cache(maxsize=1) def completion_meta_type(): # Public C registration API만 호출하며 metadata pointer는 Python에서 해석하지 않는다. import ctypes function = ctypes.CDLL('libnvds_meta.so').nvds_get_user_meta_type function.argtypes = [ctypes.c_char_p] function.restype = ctypes.c_int return function(b'CONTEXTRA.RFDETR.FRAME_COMPLETED.V1') def copy_completion(frame, component_id): matches = [] for meta in frame.user_meta_items(completion_meta_type()): value = meta.get_user_data_json() if not isinstance(value, dict): raise NativeMetadataError('COMPLETION_INVALID') if value.get('component') == component_id: matches.append(value) if not matches: raise NativeMetadataError('NATIVE_NOT_COMPLETED') if len(matches) != 1: raise NativeMetadataError('COMPLETION_INVALID') value = matches[0] if (type(value.get('schema')) is not int or value['schema'] != 1 or type(value.get('component')) is not int or type(value.get('frame')) is not int or type(value.get('pts')) is not int or value.get('status') != 'completed' or value.get('frame') != int(frame.frame_number) or value.get('pts') != int(frame.buffer_pts) or type(value.get('count')) is not int or not 0 <= value['count'] <= 300 or type(value.get('width')) is not int or value['width'] <= 0 or type(value.get('height')) is not int or value['height'] <= 0): raise NativeMetadataError('COMPLETION_INVALID') return value class NativeMetadataError(ValueError): """민감한 SDK 예외 문자열 대신 고정 reason과 크기만 전달한다.""" def __init__(self, reason, dimensions=None): super().__init__(reason) self.reason = reason self.dimensions = dimensions def failure_diagnostic(error): reasons = {'COMPONENT_ID_INVALID', 'DIMENSIONS_UNAVAILABLE', 'DETECTION_INVALID', 'BBOX_INVALID', 'BBOX_EMPTY', 'COMPLETION_INVALID', 'NATIVE_NOT_COMPLETED'} if isinstance(error, NativeMetadataError) and error.reason in reasons: result = {'reason': error.reason} if error.dimensions is not None: result['dimensions'] = dict(zip(('source_width', 'source_height', 'pipeline_width', 'pipeline_height'), error.dimensions)) return result kind = type(error).__name__ return {'reason': 'SDK_ACCESS_OR_CONVERSION_FAILED', 'error_type': kind if kind in {'AttributeError', 'TypeError', 'ValueError', 'OverflowError', 'RuntimeError'} else 'OTHER'} @dataclass(frozen=True) class NativeDetection: class_id: int score: float bbox_xyxy: tuple[float, float, float, float] keypoints: tuple[tuple[float, float, float], ...] | None = None def downstream(self): x1, y1, x2, y2 = self.bbox_xyxy return {'class_id': self.class_id, 'score': self.score, 'box': ((x1+x2)/2, (y1+y2)/2, x2-x1, y2-y1)} @dataclass(frozen=True) class NativeFrame: frame_number: int pts: int width: int height: int detections: tuple[NativeDetection, ...] bbox_width: int bbox_height: int @lru_cache(maxsize=1) def pose_meta_type(): import ctypes function = ctypes.CDLL('libnvds_meta.so').nvds_get_user_meta_type function.argtypes = [ctypes.c_char_p] function.restype = ctypes.c_int return function(b'CONTEXTRA.RFDETR.POSE.V1') def copy_pose_frame(frame, component_id): completion = copy_completion(frame, component_id) rows = [] for meta in frame.user_meta_items(pose_meta_type()): data = meta.get_user_data_json() if not isinstance(data, dict): raise NativeMetadataError('POSE_INVALID') if data.get('component') == component_id: rows.append(data) if len(rows) != 1: raise NativeMetadataError('POSE_INVALID') data = rows[0] if any(data.get(k) != completion[k] for k in ('schema','component','frame','pts')): raise NativeMetadataError('POSE_INVALID') poses = data.get('poses') if type(poses) is not list or len(poses) != completion['count'] or len(poses)>100: raise NativeMetadataError('POSE_INVALID') width, height = int(frame.source_width), int(frame.source_height) if min(width,height)<=0: raise NativeMetadataError('DIMENSIONS_UNAVAILABLE') detections=[] for p in poses: if type(p) is not dict or set(p) != {'box','score','joints'}: raise NativeMetadataError('POSE_INVALID') box=tuple(p['box']) joints=tuple(tuple(k) for k in p['joints']) if (len(box)!=4 or not all(isfinite(x) and 0<=x<=1 for x in box) or box[2]<=box[0] or box[3]<=box[1] or not isfinite(p['score']) or not 0<=p['score']<=1 or len(joints)!=17 or any(len(k)!=3 or not all(isfinite(x) for x in k) or not 0<=k[2]<=1 for k in joints)): raise NativeMetadataError('POSE_INVALID') detections.append(NativeDetection(0,p['score'],box,joints)) return NativeFrame(int(frame.frame_number),int(frame.buffer_pts),width,height,tuple(detections), completion['width'],completion['height']) def copy_detection_frame(frame, component_id: int) -> NativeFrame: if type(component_id) is not int or component_id <= 0: raise NativeMetadataError('COMPONENT_ID_INVALID') completion = copy_completion(frame, component_id) width, height = int(frame.source_width), int(frame.source_height) # SDK FrameMeta 복사에서 pipeline 크기 필드가 유실돼도 Native JSON은 copy/release로 보존된다. pw, ph = completion['width'], completion['height'] reported = int(frame.pipeline_width), int(frame.pipeline_height) if min(width, height) <= 0: raise NativeMetadataError('DIMENSIONS_UNAVAILABLE', (width, height, *reported)) if (reported[0] not in (0, pw) or reported[1] not in (0, ph)): raise NativeMetadataError('COMPLETION_INVALID') detections = [] # list(frame.object_items)는 SDK iterator 수명이 끝난 wrapper를 남길 수 있다. for obj in frame.object_items: if int(obj.unique_component_id) != component_id: continue class_id, score = int(obj.class_id), float(obj.confidence) rect = obj.rect_params left, top, rw, rh = (float(rect.left), float(rect.top), float(rect.width), float(rect.height)) if not (0 <= class_id < 80 and isfinite(score) and 0 <= score <= 1): raise NativeMetadataError('DETECTION_INVALID', (width, height, pw, ph)) if not all(isfinite(x) for x in (left, top, rw, rh)) or rw <= 0 or rh <= 0: raise NativeMetadataError('BBOX_INVALID', (width, height, pw, ph)) x1, y1 = max(0., min(1., left/pw)), max(0., min(1., top/ph)) x2, y2 = max(0., min(1., (left+rw)/pw)), max(0., min(1., (top+rh)/ph)) if x2 <= x1 or y2 <= y1: raise NativeMetadataError('BBOX_EMPTY', (width, height, pw, ph)) detections.append(NativeDetection(class_id, score, (x1, y1, x2, y2))) if len(detections) != completion['count']: raise NativeMetadataError('COMPLETION_INVALID') return NativeFrame(int(frame.frame_number), int(frame.buffer_pts), width, height, tuple(detections), pw, ph)