""" Contextra AI Analytics — FastAPI Application Entrypoint (v1.0) MLOps 베이스 레이어: - 헬스체크 엔드포인트 - Triton 모델 상태 조회 API - 기본 추론 API (RF-DETR on Triton) + PostgreSQL 메타데이터 저장 - Redis Streams Metadata Publisher 연동 상세 시나리오는 app/scenarios/ 플러그인으로 분리 예정 """ import os import asyncio from contextlib import asynccontextmanager from fastapi import APIRouter, FastAPI, HTTPException, Query from fastapi.responses import JSONResponse from datetime import datetime from typing import Optional from pydantic import BaseModel from app.inference_engine import engine, DEFAULT_MODEL_NAME from app.metadata_storage import storage, redis_publisher, FrameMeta, DetectionMeta, EventMeta from app.inference_config import rules_snapshot from app.runtime_health import collect_runtime_health from app.api.streams import router as streams_router, _sm from app.api.errors import install_stream_error_handlers router = APIRouter() async def _boot_streams(publisher): """저장된 Stream 설정을 복구하되 실제 Start는 요청 전까지 하지 않는다.""" from app.stream_manager import manager try: manager.configure_metadata(publisher) except Exception: raise RuntimeError('Metadata composition failed.') from None try: manager.load_and_autostart() except Exception: print('[STREAM-MGR] configuration load failed (non-fatal)', flush=True) def _runtime_health_response(): summary = collect_runtime_health( engine=engine, storage=storage, redis_publisher=redis_publisher, model_name=DEFAULT_MODEL_NAME, ) return JSONResponse(content=summary, status_code=200 if summary["ready"] else 503) @router.get("/live") async def live(): """FastAPI 프로세스가 요청에 응답 가능한지 확인한다.""" return {"status": "alive"} @router.get("/health") async def health(): """필수 Runtime dependency의 상세 상태를 반환한다.""" return _runtime_health_response() @router.get("/ready") async def ready(): """Triton model, PostgreSQL과 Redis 준비 상태를 판정한다.""" return _runtime_health_response() @router.get("/models/{model_name}") async def get_model_info(model_name: str): """Triton 모델 메타데이터 조회 (입력/출력 shape 포함)""" info = engine.get_model_info(model_name) if "error" in info: raise HTTPException(status_code=503, detail=f"Model query failed: {info['error']}") return info @router.post("/infer/{model_name}") async def infer( model_name: str, image_path: Optional[str] = Query(default=None, description="컨테이너 내부 이미지 경로"), confidence_threshold: float = Query(0.3, ge=0.0, le=1.0), stream_id: str = Query("CAM_01", description="스트림 ID (메타데이터용)"), site_id: str = Query("SITE01", description="사이트 ID (메타데이터용)"), ): """ 기본 추론 API — RF-DETR 객체 탐지 + PostgreSQL 저장 + Redis Streams 발행 Args: model_name: Triton 모델명 (예: RFDETRMedium) image_path: 컨테이너 내부 이미지 경로 예: /opt/contextra/app/test_user_images/B5-1024x576.jpg confidence_threshold: 탐지 신뢰도 임계값 (기본 0.3) stream_id: 스트림 ID (메타데이터용, 기본 CAM_01) site_id: 사이트 ID (메타데이터용, 기본 SITE01) Returns: 탐지 결과 목록 (class_id, class_name, confidence, bbox) """ if not image_path: image_path = "/opt/contextra/app/test_user_images/B5-1024x576.jpg" # 모델 READY 확인 if not engine.is_model_ready(model_name): raise HTTPException( status_code=503, detail=f"Model '{model_name}' is NOT ready on Triton ({engine.endpoint})" ) # 추론 실행 result = engine.infer( image_path=image_path, model_name=model_name, confidence_threshold=confidence_threshold ) if result.error: raise HTTPException(status_code=500, detail=result.error) now_ms = int(datetime.utcnow().timestamp() * 1000) seq_id = f"{now_ms % 100000:05d}" private_key = f"{site_id}_{stream_id.replace('CAM_', '')}_{now_ms}_{seq_id}" # PostgreSQL 메타데이터 저장 (비동기 — API 응답에 영향 없음) try: frame_meta = FrameMeta( private_key=private_key, stream_id=stream_id, site_id=site_id, timestamp_utc_ms=now_ms, inference_fps=15.0, width=1920, height=1080 ) storage.insert_frame(frame_meta) detection_metas = [] for d in result.detections: detection_metas.append(DetectionMeta( private_key=private_key, track_id=None, # TODO: tracker 연동 시 할당 class_label=d.class_name.lower().replace(" ", "_"), confidence=d.confidence, bbox=d.bbox, bbox_normalized=d.bbox_normalized )) if detection_metas: storage.insert_detections(detection_metas) except Exception as e: # DB 저장 실패가 추론 결과를 차단하지 않음 — 로그만 남김 print(f"[WARN] Metadata storage failed (non-fatal): {e}") # Redis Streams 발행 (fire-and-forget — API 응답에 영향 없음) try: redis_publisher.publish_frame(frame_meta) if detection_metas: redis_publisher.publish_detections(detection_metas) except Exception: pass # Redis 실패해도 추론 결과 반환 return { "model": model_name, "image": image_path, "private_key": private_key, "timestamp": datetime.utcnow().isoformat(), "detection_count": len(result.detections), "detections": [ { "class_id": d.class_id, "class_name": d.class_name, "confidence": round(d.confidence, 4), "bbox": list(d.bbox), "bbox_normalized": [round(x, 4) for x in d.bbox_normalized] } for d in result.detections ] } @router.get("/api/v1/models") async def list_models(): """Triton에 로드된 모든 모델 목록 (det + pose + seg)""" models = [] candidate_models = [ # det 계열 "RFDETRNano", "RFDETRBase", "RFDETRSmall", "RFDETRMedium", "RFDETRLarge", # keypoint (pose) "RFDETRKeypointPreview", # seg 계열 — 기존에 누락되어 ready 상태를 보고하지 않던 모델 5종 추가 "RFDETRSegNano", "RFDETRSegSmall", "RFDETRSegMedium", "RFDETRSegLarge", "RFDETRSegXLarge", ] for name in candidate_models: ready = engine.is_model_ready(name) models.append({"name": name, "ready": ready}) return {"loaded_models": models} @router.get("/api/v1/inference-rules") async def get_inference_rules(): """베이스 추론 룰 (역할별 기본 모델 + 폴백 체인 + 활성/강등 상태). MLOps 가시성: det/pose/seg 각각 어떤 모델이 활성인지, primary가 죽고 fallback으로 강등(degraded=true)된 상태인지 한눈에 확인. 읽기 전용 — 라이브 파이프라인 경로는 건드리지 않는다 (전환은 별도 단계).""" return rules_snapshot() @router.get("/api/v1/frames") async def get_frames(stream_id: str = "CAM_01", limit: int = 30): """클라이언트용: 최근 프레임 + 탐지 결과 조회""" try: frames = storage.get_frames_by_stream(stream_id, limit=limit) return {"stream_id": stream_id, "frames": frames} except Exception as e: raise HTTPException(status_code=503, detail=f"DB query failed: {e}") # --------------------------------------------------------------------------- # 스트림 관리 API — /api/v1/streams (추가/제거/시작/중지/재시작/상태) # 소스 타입: rtsp | hls | file | onvif | webrtc # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Zone API — 시나리오 ROI (사용자 인터랙티브 드로잉 영속화, Phase 2) # 좌표는 정규화(0~1) polygon/rect. UI가 스냅샷 위에 드로잉 → 이 API로 저장 → # 스트림 재시작 시 엔진이 zone을 로드해 침입/배회 판정에 사용. # --------------------------------------------------------------------------- class ZoneSpec(BaseModel): name: str # restricted_zone, danger_zone 등 (논리 이름) geometry: dict # {"type":"polygon"|"rect","points":[[x,y],...]} (0~1) is_active: Optional[bool] = None # upsert 시만 사용 (None=기존 유지/기본 true) def _zs(): from app.zone_store import get_zone_store, ZoneError return get_zone_store(), ZoneError @router.get("/api/v1/streams/{stream_id}/zones") async def list_stream_zones(stream_id: str): """스트림의 zone 목록 (비활성 포함).""" zs, _ = _zs() try: zones = zs.list_zones(stream_id=stream_id, include_inactive=True) except Exception as e: raise HTTPException(status_code=503, detail=f"zone query failed: {e}") return {"stream_id": stream_id, "zones": zones} @router.put("/api/v1/streams/{stream_id}/zones/{name}") async def upsert_stream_zone(stream_id: str, name: str, spec: ZoneSpec): """zone 생성/갱신 (갱신 시 version+1). geometry 검증 실패 → 422. 저장만 수행 — 적용은 해당 스트림 /restart 시 (엔진이 기동 시 zone 로드).""" zs, ZoneError = _zs() try: z = zs.upsert_zone(stream_id=stream_id, site_id="SITE01", name=name, geometry=spec.geometry, is_active=spec.is_active) except ZoneError as e: raise HTTPException(status_code=422, detail=str(e)) except Exception as e: raise HTTPException(status_code=503, detail=f"zone upsert failed: {e}") return {"stream_id": stream_id, "zone": z} @router.delete("/api/v1/streams/{stream_id}/zones/{name}") async def delete_stream_zone(stream_id: str, name: str): zs, _ = _zs() try: deleted = zs.delete_zone(stream_id=stream_id, name=name) except Exception as e: raise HTTPException(status_code=503, detail=f"zone delete failed: {e}") if not deleted: raise HTTPException(status_code=404, detail=f"zone {name} not found for {stream_id}") return {"stream_id": stream_id, "deleted": name} @router.post("/api/v1/streams/{stream_id}/zones/{name}/active") async def set_zone_active(stream_id: str, name: str, active: bool = True): """zone 활성 토글 (삭제 없이 일시 비활성화).""" zs, _ = _zs() try: changed = zs.set_active(stream_id=stream_id, name=name, is_active=active) except Exception as e: raise HTTPException(status_code=503, detail=f"zone update failed: {e}") if not changed: raise HTTPException(status_code=404, detail=f"zone {name} not found for {stream_id}") return {"stream_id": stream_id, "name": name, "is_active": active} # --------------------------------------------------------------------------- # 이벤트 검증 API — 오탐/정답 라벨링 + 증거 프레임 조회 # persist_mode=event_gated 스트림은 이벤트 ±윈도우 프레임만 PG에 저장하므로, # 이 엔드포인트로 이벤트 목록 → 증거 프레임 확인 → is_verified 라벨링 워크플로우. # --------------------------------------------------------------------------- def _storage(): from app.metadata_storage import storage return storage @router.get("/api/v1/events") async def list_events( stream_id: str | None = None, event_type: str | None = None, verified: str | None = None, # 'true'/'false' → bool (기타 값 무시) limit: int = 50, offset: int = 0, ): """이벤트 목록 조회 — context_frames(증거 프레임 수) 포함. verified 필터: true=검증 완료만, false=미검증만. """ is_verified = None if verified == "true": is_verified = True elif verified == "false": is_verified = False limit = max(1, min(int(limit), 200)) offset = max(0, int(offset)) try: events = _storage().list_events(stream_id=stream_id, event_type=event_type, is_verified=is_verified, limit=limit, offset=offset) except Exception as e: raise HTTPException(status_code=503, detail=f"storage unavailable: {e}") return {"events": events, "limit": limit, "offset": offset} @router.get("/api/v1/events/{event_id}") async def get_event(event_id: str): """단일 이벤트 상세 (증거 프레임 수 포함).""" try: ev = _storage().get_event(event_id) except Exception as e: raise HTTPException(status_code=503, detail=f"storage unavailable: {e}") if ev is None: raise HTTPException(status_code=404, detail=f"event {event_id} not found") return ev @router.get("/api/v1/events/{event_id}/frames") async def get_event_frames(event_id: str, limit: int = 100): """이벤트 증거 프레임 목록 — frames.event_context->>'event_id' 매칭. 이벤트 게이트 영속화 시 저장된 프레임은 전부 이 이벤트의 ±윈도우(전 10s/후 5s) 증거. """ limit = max(1, min(int(limit), 500)) try: ev = _storage().get_event(event_id) if ev is None: raise HTTPException(status_code=404, detail=f"event {event_id} not found") frames = _storage().get_event_frames(event_id, limit=limit) except HTTPException: raise except Exception as e: raise HTTPException(status_code=503, detail=f"storage unavailable: {e}") return {"event_id": event_id, "frames": frames} class EventVerifyRequest(BaseModel): is_verified: bool = True # true=정답(실 이벤트), false=오탐 verified_by: str = "api" @router.post("/api/v1/events/{event_id}/verify") async def verify_event(event_id: str, body: EventVerifyRequest): """이벤트 검증 라벨링 — 오탐(false)/정답(true) 구분. is_verified=false → 오탐(허위 경보), true → 정답(실제 이벤트). verified_at/verified_by 자동 기록. """ try: ok = _storage().verify_event(event_id, body.is_verified, body.verified_by) if not ok and _storage().get_event(event_id) is None: raise HTTPException(status_code=404, detail=f"event {event_id} not found") ev = _storage().get_event(event_id) except HTTPException: raise except Exception as e: raise HTTPException(status_code=503, detail=f"storage unavailable: {e}") return ev # --------------------------------------------------------------------------- # Prometheus /metrics — MLOps 관측성 (베이스 레이어) # 스트림 상태 / 프레임 처리량 / Triton 모델 ready / 프로세스 uptime 노출. # prometheus_client는 이미 설치됨. 별도 프로세스 없이 FastAPI 엔드포인트로 # 노출 → docker/prometheus.yml이 스크레이프. # --------------------------------------------------------------------------- import time as _time_mod from fastapi.responses import Response as _PromResponse from prometheus_client import ( CollectorRegistry, Counter, Gauge, generate_latest, CONTENT_TYPE_LATEST, ) _APP_START_TS = _time_mod.time() _metrics_registry = CollectorRegistry() _m_stream_state = Gauge( "contextra_stream_state", "Stream state gauge (1=running, 0=stopped/error)", ["stream_id"], registry=_metrics_registry) _m_frames_window = Gauge( "contextra_stream_frames_in_window", "Frames observed in recent window per stream", ["stream_id"], registry=_metrics_registry) _m_triton_ready = Gauge( "contextra_triton_model_ready", "Triton model ready (1/0)", ["model"], registry=_metrics_registry) _m_uptime_seconds = Gauge( "contextra_process_uptime_seconds", "FastAPI process uptime in seconds", registry=_metrics_registry) _m_frames_pruned = Counter( "contextra_frames_pruned_total", "Non-evidence frames pruned by retention policy", registry=_metrics_registry) _METRIC_MODELS = [ "RFDETRNano", "RFDETRBase", "RFDETRSmall", "RFDETRMedium", "RFDETRLarge", "RFDETRKeypointPreview", "RFDETRSegNano", "RFDETRSegSmall", "RFDETRSegMedium", "RFDETRSegLarge", "RFDETRSegXLarge", ] @router.get("/metrics") async def prometheus_metrics(): """Prometheus text-format metrics — live state from stream manager + Triton.""" try: _m_uptime_seconds.set(_time_mod.time() - _APP_START_TS) except Exception: pass for name in _METRIC_MODELS: try: _m_triton_ready.labels(model=name).set(1.0 if engine.is_model_ready(name) else 0.0) except Exception: _m_triton_ready.labels(model=name).set(0.0) try: manager, _ = _sm() for s in manager.list(detail=False): sid = str(s.get("stream_id", "unknown")) running = 1.0 if s.get("state") == "running" else 0.0 _m_stream_state.labels(stream_id=sid).set(running) try: _m_frames_window.labels(stream_id=sid).set(float(s.get("frames_in_window", 0))) except Exception: pass except Exception: pass return _PromResponse(content=generate_latest(_metrics_registry), media_type=CONTENT_TYPE_LATEST) def _boot_retention(): """frames retention — N일 초과 비증거 프레임 주기적 삭제 (daemon 스레드). FRAME_RETENTION_DAYS env (기본 30). 이벤트 증거 프레임(event_context)은 보존. 기동 60s 후 첫 실행(스트림 autostart/DB 안정 대기), 이후 6시간 주기.""" import threading from app.metadata_storage import storage as _st days = int(os.environ.get("FRAME_RETENTION_DAYS", "30")) stopped = threading.Event() def _loop(): while not stopped.is_set(): try: n = _st.prune_old_frames(retention_days=days) if n: print(f"[RETENTION] pruned {n} non-evidence frames older than {days}d", flush=True) try: _m_frames_pruned.inc(n) except Exception: pass except Exception: print("[RETENTION] prune failed (non-fatal)", flush=True) if stopped.wait(6 * 3600): return def _delayed(): if not stopped.wait(60): _loop() worker = threading.Thread(target=_delayed, name="frame-retention", daemon=True) worker.start() return stopped, worker @asynccontextmanager async def _lifespan(application): # COMPOSITION-01: router에는 event를 등록하지 않고 앱이 준비한 자원만 회수한다. from app.metadata_runtime import publisher_from_environment publisher = None retention = None artifacts = None training = None failed = False application.state.cleanup_errors = () try: if os.environ.get('CONTEXTRA_METADATA_VISUAL_DIR'): from app.capture_pixels import prepare_capture application.state.capture_preparation = prepare_capture() publisher = publisher_from_environment() application.state.metadata_publisher = publisher from app.event_artifact_service import ArtifactService artifacts = ArtifactService.from_environment() if artifacts is not None: await asyncio.to_thread(artifacts.start) application.state.event_artifacts = artifacts from app.training_store import root_path from app.training_service import TrainingService if root_path() is not None: training = TrainingService(root_path(), artifacts) await asyncio.to_thread(training.start) application.state.training = training await _boot_streams(publisher) retention = _boot_retention() application.state.retention = retention yield except BaseException: failed = True raise finally: errors = [] from app.stream_manager import manager try: if not await asyncio.to_thread(manager.shutdown_channels): errors.append('CHANNEL_PROCESS_CLEANUP_INCOMPLETE') except Exception: errors.append('CHANNEL_PROCESS_CLEANUP_FAILED') if training is not None: try: if not await asyncio.to_thread(training.close): errors.append('TRAINING_CLEANUP_INCOMPLETE') except Exception: errors.append('TRAINING_CLEANUP_FAILED') if artifacts is not None: try: if not await asyncio.to_thread(artifacts.close): errors.append('ARTIFACT_CLEANUP_INCOMPLETE') except Exception: errors.append('ARTIFACT_CLEANUP_FAILED') if retention is not None: stopped, worker = retention stopped.set() try: await asyncio.to_thread(worker.join, 5) if worker.is_alive(): errors.append('RETENTION_CLEANUP_INCOMPLETE') except Exception: errors.append('RETENTION_CLEANUP_FAILED') if publisher is not None: try: if not await asyncio.to_thread(publisher.close, 1): errors.append('METADATA_CLEANUP_INCOMPLETE') except Exception: errors.append('METADATA_CLEANUP_FAILED') application.state.cleanup_errors = tuple(errors) if errors: print('[LIFESPAN] ' + ','.join(errors), flush=True) # 초기 실패를 cleanup 오류로 덮지 않고 참조와 별도 상태를 남긴다. if not failed: raise RuntimeError('Application cleanup incomplete.') from None def create_app() -> FastAPI: """Build a complete ASGI application without running startup side effects.""" # COMPOSITION-01: FastAPI Route와 startup handler 조립은 이 factory가 단독 소유한다. application = FastAPI(title="Contextra Analytics API", version="1.1.0", lifespan=_lifespan) application.include_router(router) application.include_router(streams_router) install_stream_error_handlers(application) return application app = create_app() # --------------------------------------------------------------------------- # CLI entry point — python -m app.main [--port 5000] # --------------------------------------------------------------------------- if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Contextra FastAPI server") parser.add_argument("--port", type=int, default=5000, help="Listen port (default: 5000)") args = parser.parse_args() print(f"[FastAPI] Starting uvicorn on port {args.port} ...") import uvicorn uvicorn.run( app, host="0.0.0.0", port=args.port, log_level="info", )