""" Contextra AI Analytics — Triton Inference Engine (v2.0) MLOps 베이스 레이어: GPU 추론 헬스체크 + RF-DETR 추론 API RF-DETR TensorRT output format: dets : [cx, cy, w, h] normalized to input_size (0~1 range) labels : raw logits (COCO 91 classes including background at index 0) Postprocessing pipeline: 1. Shared parser (rfdetr_postprocess.parse_boxes_labels): sigmoid + background(0) mask + per-query argmax + score > threshold 2. Filter degenerate boxes (w>1 & h>1 px) 3. Convert cxcywh to xyxy, scale to original image size 4. model background 제외 → canonical zero-based class_id (2026-08-17: inline 구현을 공용 파서 호출로 교체 — RTSP/API 동일 코드 경로) 역할: analytics 컨테이너에서 Triton 컨테이너로 gRPC 요청 전송 시나리오 로직은 별도 app/scenarios/ 플러그인으로 확장 """ import os import numpy as np from PIL import Image import tritonclient.grpc as grpcclient from tritonclient.utils import InferenceServerException # 공용 RF-DETR 포스트프로세서 (RTSP 파이프라인과 동일 코드 경로) from app.rfdetr_postprocess import parse_boxes_labels, squeeze_batch from dataclasses import dataclass, field from typing import Optional # COCO 80 classes (index 0 = background) COCO_CLASSES = [ "__background__", "person", "bicycle", "car", "motorcycle", "airplane", "bus", "train", "truck", "boat", "traffic light", "fire hydrant", "N/A", "stop sign", "parking meter", "bench", "bird", "cat", "dog", "horse", "sheep", "cow", "elephant", "bear", "zebra", "giraffe", "N/A", "backpack", "umbrella", "N/A", "N/A", "handbag", "tie", "suitcase", "frisbee", "skis", "snowboard", "sports ball", "kite", "baseball bat", "baseball glove", "skateboard", "surfboard", "tennis racket", "bottle", "N/A", "wine glass", "cup", "fork", "knife", "spoon", "bowl", "banana", "apple", "sandwich", "orange", "broccoli", "carrot", "hot dog", "pizza", "donut", "cake", "chair", "couch", "potted plant", "bed", "N/A", "dining table", "N/A", "N/A", "toilet", "N/A", "tv", "laptop", "mouse", "remote", "keyboard", "cell phone", "microwave", "oven", "toaster", "sink", "refrigerator", "N/A", "book", "clock", "vase", "scissors", "teddy bear", "hair drier", "toothbrush" ] @dataclass class Detection: """RF-DETR 탐지 결과""" class_id: int # COCO foreground class index (0-based, maps to COCO_CLASSES[1:]) class_name: str # Human-readable name confidence: float # Sigmoid probability [0.0, 1.0] bbox: tuple # (x_min, y_min, x_max, y_max) in original image pixel coords bbox_normalized: tuple # (x_min/w, y_min/h, x_max/w, y_max/h) relative to original image @dataclass class InferenceResult: """추론 결과 래퍼""" model_name: str detections: list = field(default_factory=list) inference_time_ms: float = 0.0 error: Optional[str] = None def _box_cxcywh_to_xyxy(boxes_cxcywh, orig_w, orig_h): """ Convert normalized [cx, cy, w, h] to pixel [x_min, y_min, x_max, y_max]. RF-DETR outputs boxes in cxcywh format normalized to model input size. We need to scale them back to the original image dimensions. """ cx = np.clip(boxes_cxcywh[:, 0], 0, 1) * orig_w cy = np.clip(boxes_cxcywh[:, 1], 0, 1) * orig_h w = np.clip(boxes_cxcywh[:, 2], 0, 1) * orig_w h = np.clip(boxes_cxcywh[:, 3], 0, 1) * orig_h x_min = cx - w / 2.0 y_min = cy - h / 2.0 x_max = cx + w / 2.0 y_max = cy + h / 2.0 return np.stack([x_min, y_min, x_max, y_max], axis=-1) def _postprocess_rfdetr(dets_raw, labels_raw, orig_w, orig_h, confidence_threshold=0.3): """ RF-DETR TensorRT postprocessing — 공용 파서(rfdetr_postprocess) 위임. 선택 코어(.sigmoid, bg(0) 마스킹, per-query argmax, score > threshold)는 RTSP 파이프라인과 동일한 rfdetr_postprocess.parse_boxes_labels를 사용. API 경로 특유의 후반부 처리만 이 함수에서 수행: - 퇴화 박스 필터 (w>1 & h>1 px) - cxcywh 정규화 → 원본 픽셀 xyxy 변환 - canonical zero-based class_id와 class name 매핑 Args: dets_raw: numpy array [batch, Q, 4] — normalized cxcywh boxes labels_raw: numpy array [batch, Q, C] — raw logits (COCO 91 classes) orig_w: original image width in pixels orig_h: original image height in pixels confidence_threshold: minimum probability to keep detection Returns: List of Detection dataclass instances """ dets = squeeze_batch(dets_raw, 2) labels = squeeze_batch(labels_raw, 2) parsed = parse_boxes_labels(dets, labels, threshold=confidence_threshold) classes = parsed["classes"] # [Q] — canonical zero-based foreground index scores = parsed["scores"] # [Q] keep = parsed["keep"].astype(bool) # cxcywh(정규화) → 원본 이미지 픽셀 xyxy boxes_xyxy = _box_cxcywh_to_xyxy(parsed["dets"], orig_w, orig_h) # 퇴화 박스 필터 (legacy와 동일) box_widths = np.clip(boxes_xyxy[:, 2] - boxes_xyxy[:, 0], 0, None) box_heights = np.clip(boxes_xyxy[:, 3] - boxes_xyxy[:, 1], 0, None) keep &= (box_widths > 1.0) & (box_heights > 1.0) # Build detection results detections = [] for idx in np.where(keep)[0]: cls_id_internal = int(classes[idx]) conf = float(scores[idx]) model_class_id = cls_id_internal + 1 bbox_pixels = tuple(int(round(v)) for v in boxes_xyxy[idx]) bbox_norm = ( float(boxes_xyxy[idx, 0] / orig_w), float(boxes_xyxy[idx, 1] / orig_h), float(boxes_xyxy[idx, 2] / orig_w), float(boxes_xyxy[idx, 3] / orig_h) ) cls_name = ( COCO_CLASSES[model_class_id] if 0 < model_class_id < len(COCO_CLASSES) else f"unknown_{cls_id_internal}" ) detections.append(Detection( class_id=cls_id_internal, class_name=cls_name, confidence=round(conf, 4), bbox=bbox_pixels, bbox_normalized=tuple(round(v, 4) for v in bbox_norm) )) return detections # Triton gRPC 엔드포인트 (환경변수에서 읽음, 기본값은 docker compose 네트워크 기준) TRITON_GRPC_ENDPOINT = os.environ.get( "TRITON_GRPC_ENDPOINT", "contextra-triton:8001" ) # 모델 설정 DEFAULT_MODEL_NAME = os.environ.get("MODEL_NAME", "RFDETRMedium") MODEL_PRECISION = os.environ.get("MODEL_PRECISION", "fp16") MODEL_RESOLUTION = tuple( int(x) for x in os.environ.get("MODEL_RESOLUTION", "576x576").split("x") ) class TritonInferenceEngine: """Triton Inference Server gRPC 클라이언트""" def __init__(self, endpoint: str = TRITON_GRPC_ENDPOINT): self.endpoint = endpoint self.client = grpcclient.InferenceServerClient(url=endpoint) self._model_ready_cache: dict[str, bool] = {} def is_model_ready(self, model_name: str, force_refresh: bool = False) -> bool: """모델이 READY인지 확인 (캐싱됨)""" if not force_refresh and model_name in self._model_ready_cache: return self._model_ready_cache[model_name] try: is_ready = self.client.is_model_ready(model_name) self._model_ready_cache[model_name] = is_ready return is_ready except Exception as e: print(f"[WARN] Model readiness check failed for {model_name}: {e}") self._model_ready_cache[model_name] = False return False def get_model_info(self, model_name: str) -> dict: """모델 메타데이터 조회 (입력/출력 shape 포함)""" try: metadata = self.client.get_model_config( model_name=model_name, as_json=True ) return { "name": model_name, "inputs": [ {"name": i["name"], "shape": list(i.get("dims", [])), "datatype": i["data_type"]} for i in metadata.get("config", {}).get("input", []) ], "outputs": [ {"name": o["name"], "shape": list(o.get("dims", [])), "datatype": o["data_type"]} for o in metadata.get("config", {}).get("output", []) ] } except Exception as e: return {"error": str(e)} def infer( self, image_path: str, model_name: str = DEFAULT_MODEL_NAME, confidence_threshold: float = 0.3, resize_to: tuple = MODEL_RESOLUTION ) -> InferenceResult: """ 이미지 추론 실행 Args: image_path: 로컬 파일 경로 (컨테이너 내부 기준) model_name: Triton 모델명 confidence_threshold: 탐지 신뢰도 임계값 resize_to: 추론 입력 크기 (width, height) Returns: InferenceResult 객체 """ result = InferenceResult(model_name=model_name) # 1. 모델 READY 확인 if not self.is_model_ready(model_name): result.error = f"Model '{model_name}' is NOT ready on Triton ({self.endpoint})" return result # 2. 이미지 로드 및 전처리 try: img = Image.open(image_path).convert("RGB") orig_w, orig_h = img.size arr = np.array(img.resize(resize_to)).astype(np.float32) / 255.0 batch = np.expand_dims(arr.transpose(2, 0, 1), axis=0).astype(np.float32) except Exception as e: result.error = f"Image load failed: {e}" return result # 3. Triton 추론 요청 try: inputs = [grpcclient.InferInput("input", batch.shape, "FP32")] inputs[0].set_data_from_numpy(batch) response = self.client.infer( model_name=model_name, inputs=inputs ) except InferenceServerException as e: result.error = f"Triton inference failed: {e}" return result except Exception as e: result.error = f"gRPC/Inference error: {e}" return result # 4. RF-DETR postprocessing (sigmoid + cxcywh→xyxy + COCO class mapping) try: dets_raw = response.as_numpy("dets") # [1, Q, 4] or [Q, 4] labels_raw = response.as_numpy("labels") # [1, Q, C] or [Q, C] result.detections = _postprocess_rfdetr( dets_raw=dets_raw, labels_raw=labels_raw, orig_w=orig_w, orig_h=orig_h, confidence_threshold=confidence_threshold ) except Exception as e: import traceback result.error = f"Result parsing failed: {e}\n{traceback.format_exc()}" return result return result # 전역 인스턴스 (싱글톤) engine = TritonInferenceEngine()