# app/core/config.py # -*- coding: utf-8 -*- """ Configuration settings for the inpainting server - Jetson/x86 자동 감지 - 프레임워크 충돌 방지 정책 (기본: TORCH_GPU_ONLY) - VRAM/플랫폼 기반 자동 튜닝 및 유효성 검사 - 환경변수 가드/스모크 테스트 훅 제공 """ import os import json import platform import logging from pathlib import Path from typing import Optional, ClassVar from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic import Field, AliasChoices logger = logging.getLogger(__name__) class Settings(BaseSettings): # pydantic v2 설정: .env 읽기 + 여분 키 무시 model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", extra="ignore", # .env에 있는 미정의 키는 무시 case_sensitive=True, ) # ========================= # 플랫폼/프로젝트 경로 # ========================= IS_JETSON: ClassVar[bool] = ( "aarch64" in platform.machine().lower() and "tegra" in platform.release().lower() ) PROJECT_ROOT: ClassVar[str] = os.path.dirname( os.path.dirname(os.path.dirname(os.path.abspath(__file__))) ) # ========================= # 앱 기본 # ========================= APP_VERSION: str = "3.0.0-dynamic-pool" APP_NAME: str = "Inpaint & RemoveBG Server" API_PREFIX: str = "/api/v1" HOST: str = "0.0.0.0" PORT: int = 8008 WORKERS: int = 1 # GPU 서버는 프로세스 1 권장(내부 동시성/세션풀 활용) # ========================= # GPU/프레임워크 정책 # ========================= # NOTE: .env의 GPU_DEVICE도 받아들이도록 별칭 설정 CUDA_DEVICE: int = Field(0, validation_alias=AliasChoices("CUDA_DEVICE", "GPU_DEVICE")) USE_CUDA: bool = True USE_FP16: bool = True FP16_ENABLED: bool = True # 이전 호환 # 프레임워크 사용 정책: # - TORCH_GPU_ONLY: Torch만 GPU, TF/ONNX는 CPU # - ORT_GPU: ONNXRuntime GPU, TF는 CPU # - TF_GPU_ONLY: TF만 GPU(권장 X; Torch와 혼용 시 충돌↑) FRAMEWORK_POLICY: str = "TORCH_GPU_ONLY" # ONNX Runtime / TensorRT / TensorFlow ORT_ENABLED: bool = True ORT_PROVIDERS: str = "CPUExecutionProvider" # 기본은 CPU EP (정책에 따라 변경) USE_TENSORRT: bool = False # 내부 사용 플래그 TENSORRT_FP16: bool = True TENSORRT_WORKSPACE_SIZE: int = 2 * 1024 * 1024 * 1024 # 2GB TENSORRT_ENABLED: Optional[bool] = None # .env 호환용 (명시되면 USE_TENSORRT와 동기화) TF_ENABLED: bool = False # 정책에서 최종 보정 # ========================= # Jetson 설정 # ========================= JETSON_MODE: bool = IS_JETSON JETSON_POWER_MODE: str = "MAXN" # MAXN, 5W, 10W, 15W JETSON_FAN_CONTROL: bool = True JETSON_TEMP_THRESHOLD: int = 75 # Celsius JETSON_GPU_FREQ: int = 1200 # MHz JETSON_CPU_FREQ: int = 1900 # MHz JETSON_MEMORY_FREQ: int = 1600 # MHz # ========================= # 동적 세션 풀/메모리 # ========================= SIMPLE_LAMA_MIN_SESSIONS: int = 2 SIMPLE_LAMA_MAX_SESSIONS: int = 4 # x86에서는 MIGAN 미로딩(지연 로딩) 기본 → MIN=0 MIGAN_MIN_SESSIONS: int = 2 if IS_JETSON else 1 MIGAN_MAX_SESSIONS: int = 4 REMBG_MIN_SESSIONS: int = 2 if IS_JETSON else 1 REMBG_MAX_SESSIONS: int = 4 if IS_JETSON else 3 # 여유 VRAM 비율(남은 VRAM이 이 값보다 커야 세션 추가) SESSION_VRAM_THRESHOLD: float = 0.30 SESSION_IDLE_TIMEOUT: int = 1800 # 초 (0이면 비활성) # 마이크로 배치(SimpleLAMA) USE_MICRO_BATCHING: bool = True MICRO_BATCH_SIZE: int = 4 MICRO_BATCH_TIMEOUT_MS: int = 100 # 사전 확정 세션(플랫폼 감안 기본치) SIMPLE_LAMA_SESSIONS: int = 4 MIGAN_SESSIONS: int = 4 REMBG_SESSIONS: int = 3 if IS_JETSON else 2 # 워커(내부 큐/스레드 워커, 프로세스는 WORKERS) MAX_WORKERS: int = 4 if IS_JETSON else 8 MIN_WORKERS: int = 1 if IS_JETSON else 4 WORKER_TIMEOUT: int = 120 # ========================= # 메모리/VRAM 관리 # ========================= VRAM_THRESHOLD_HIGH: float = 0.70 if IS_JETSON else 0.80 VRAM_THRESHOLD_LOW: float = 0.30 if IS_JETSON else 0.40 VRAM_CHECK_INTERVAL: int = 20 if IS_JETSON else 15 # 초 # ========================= # 모델/경로 # ========================= SIMPLE_LAMA_MODEL_PATH: str = "app/models/pt/big-lama.pt" MIGAN_MODEL_PATH: str = "app/models/onnx/migan_pipeline_v2.onnx" REMBG_MODEL_PATH: str = "app/models/onnx/BriaRMBG1.4_model_fp16.onnx" # MIGAN ONNX MIGAN_ONNX_PATH: Optional[str] = "app/models/onnx/migan_pipeline_v2.onnx" MIGAN_INTRA_THREADS: int = 0 MIGAN_INTER_THREADS: int = 0 # REMBG (자동 다운로드) REMBG_MODEL_NAME: str = "briaai/RMBG-1.4" LOCAL_REMBG_MODEL_PATH: Optional[str] = None # ========================= # 업로드/이미지 제한 # ========================= MAX_FILE_SIZE: int = (100 * 1024 * 1024) if IS_JETSON else (50 * 1024 * 1024) MAX_IMAGE_SIZE: int = 4096 if IS_JETSON else 3072 # ALLOWED_EXTENSIONS는 env에서 문자열로 받고 내부 파싱 ALLOWED_EXTENSIONS_RAW: str = Field( ".jpg,.jpeg,.png,.bmp,.tiff,.webp", validation_alias=AliasChoices("ALLOWED_EXTENSIONS", "ALLOWED_EXTENSIONS_RAW"), ) @property def ALLOWED_EXTENSIONS(self) -> set[str]: """ .env에서 다음 형식을 모두 허용: 1) JSON 리스트: [".png",".jpg"] 2) 콤마 문자열: .png,.jpg,.webp 공백/중복은 자동 정리 """ s = (self.ALLOWED_EXTENSIONS_RAW or "").strip() if not s: return {".jpg", ".jpeg", ".png", ".bmp", ".tiff"} # JSON 리스트 우선 try: data = json.loads(s) if isinstance(data, (list, tuple, set)): return {str(x).strip() for x in data if str(x).strip()} except Exception: pass # 콤마 문자열 return {p.strip() for p in s.split(",") if p.strip()} # ========================= # Jetson 부가 토글 (.env 호환) # ========================= JETSON_OPTIMIZE_ON_STARTUP: bool = False JETSON_AUTO_FAN_CONTROL: bool = False JETSON_POWER_SAVING: bool = False # ========================= # 모니터링/알림/로깅 # ========================= ENABLE_MONITORING: bool = True MONITORING_PORT: int = 8888 DISCORD_WEBHOOK_URL: Optional[str] = None LOG_LEVEL: str = "INFO" LOG_DIR: str = "logs" # ------------------------- # 런타임 보정/검증 # ------------------------- def finalize(self) -> "Settings": """ - 정책 기반 프레임워크 토글/EP 보정 - 세션 min/max/고정값 일관성 보정 - x86/Jetson별 기본값 자동 조정 - 경계/오타/모순값 사전 차단 """ # 0) policy 정상화 + TF 기본값 policy = (self.FRAMEWORK_POLICY or "TORCH_GPU_ONLY").upper() if policy not in {"TORCH_GPU_ONLY", "ORT_GPU", "TF_GPU_ONLY"}: logger.warning(f"알 수 없는 FRAMEWORK_POLICY='{self.FRAMEWORK_POLICY}', TORCH_GPU_ONLY로 강제") policy = "TORCH_GPU_ONLY" self.FRAMEWORK_POLICY = policy self.TF_ENABLED = (policy == "TF_GPU_ONLY") # 1) .env의 TENSORRT_ENABLED가 명시되면 내부 USE_TENSORRT와 동기화 if self.TENSORRT_ENABLED is not None: self.USE_TENSORRT = bool(self.TENSORRT_ENABLED) # 2) 정책별 기본 EP/TF/ORT/TPU 정리 if policy == "TORCH_GPU_ONLY": self.ORT_ENABLED = True self.ORT_PROVIDERS = "CPUExecutionProvider" self.USE_TENSORRT = False elif policy == "ORT_GPU": self.ORT_ENABLED = True if self.ORT_PROVIDERS.strip() in {"", "CPUExecutionProvider"}: # x86 기본 CUDA EP, Jetson 특수 빌드면 TensorrtExecutionProvider를 직접 넣어도 됨 self.ORT_PROVIDERS = "CUDAExecutionProvider,CPUExecutionProvider" elif policy == "TF_GPU_ONLY": self.ORT_ENABLED = False # TF-TRT 경로를 따로 준비하지 않았다면 False 유지 self.USE_TENSORRT = False # 3) 세션 min/max 일관성 보정 def fix_min_max(name_min, name_max): mn = getattr(self, name_min) mx = getattr(self, name_max) if mn < 0: logger.warning(f"{name_min} < 0 → 0으로 보정") mn = 0 if mx < mn: logger.warning(f"{name_max} < {name_min} → {name_min}로 보정") mx = mn setattr(self, name_min, mn) setattr(self, name_max, mx) fix_min_max("SIMPLE_LAMA_MIN_SESSIONS", "SIMPLE_LAMA_MAX_SESSIONS") fix_min_max("MIGAN_MIN_SESSIONS", "MIGAN_MAX_SESSIONS") fix_min_max("REMBG_MIN_SESSIONS", "REMBG_MAX_SESSIONS") # x86에서 MIGAN 지연 로딩 기본화 if not self.IS_JETSON and self.MIGAN_MIN_SESSIONS == 0 and self.MIGAN_SESSIONS > 0: logger.info("x86에서 MIGAN_MIN_SESSIONS=0 → MIGAN_SESSIONS=0(지연 로딩)") self.MIGAN_SESSIONS = 0 # 4) VRAM 임계/체크 주기 보정 if not (0.0 < self.VRAM_THRESHOLD_LOW < self.VRAM_THRESHOLD_HIGH < 0.99): logger.warning( f"VRAM 임계값 비정상(HIGH={self.VRAM_THRESHOLD_HIGH}, LOW={self.VRAM_THRESHOLD_LOW}) → 플랫폼 기본으로 보정" ) if self.IS_JETSON: self.VRAM_THRESHOLD_HIGH, self.VRAM_THRESHOLD_LOW = 0.70, 0.30 self.VRAM_CHECK_INTERVAL = 20 else: self.VRAM_THRESHOLD_HIGH, self.VRAM_THRESHOLD_LOW = 0.80, 0.40 self.VRAM_CHECK_INTERVAL = 15 # 5) 파일/이미지 제한 sanity check if self.MAX_FILE_SIZE <= 0: self.MAX_FILE_SIZE = (100 * 1024 * 1024) if self.IS_JETSON else (50 * 1024 * 1024) if self.MAX_IMAGE_SIZE <= 0: self.MAX_IMAGE_SIZE = 4096 if self.IS_JETSON else 3072 # 6) REMBG 모델 문자열 교정 if self.REMBG_MODEL_NAME.replace(" ", "") in {"briaaiRMBG-1.4", "briaai\\RMBG-1.4"}: self.REMBG_MODEL_NAME = "briaai/RMBG-1.4" # 7) 로그 안내 logger.info( f"플랫폼: {'Jetson' if self.IS_JETSON else 'x86_64'}, 정책: {self.FRAMEWORK_POLICY}, " f"ORT_PROVIDERS: {self.ORT_PROVIDERS}, TF_ENABLED: {self.TF_ENABLED}" ) logger.info( f"세션(MIN~MAX/boot): " f"LAMA {self.SIMPLE_LAMA_MIN_SESSIONS}~{self.SIMPLE_LAMA_MAX_SESSIONS}/{self.SIMPLE_LAMA_SESSIONS}, " f"MIGAN {self.MIGAN_MIN_SESSIONS}~{self.MIGAN_MAX_SESSIONS}/{self.MIGAN_SESSIONS}, " f"REMBG {self.REMBG_MIN_SESSIONS}~{self.REMBG_MAX_SESSIONS}/{self.REMBG_SESSIONS}" ) logger.info( f"VRAM 임계: HIGH={self.VRAM_THRESHOLD_HIGH:.2f}, LOW={self.VRAM_THRESHOLD_LOW:.2f}, " f"CHECK_INTERVAL={self.VRAM_CHECK_INTERVAL}s" ) return self # ------------------------- # 전역 설정 인스턴스 # ------------------------- settings = Settings().finalize() # 파일에서 웹훅 URL 로드(환경 변수보다 우선순위 낮음) if not settings.DISCORD_WEBHOOK_URL: try: webhook_file = Path(Settings.PROJECT_ROOT) / "webhook_url.txt" if webhook_file.exists(): url = webhook_file.read_text().strip() if url: settings.DISCORD_WEBHOOK_URL = url logger.info(f"파일에서 Discord 웹훅 URL을 로드했습니다: {url[:30]}...") except Exception as e: logger.warning(f"webhook_url.txt 파일 로드 실패: {e}") # ------------------------- # 런타임 가드/스모크 테스트 헬퍼 # ------------------------- def apply_env_guards() -> None: """ 프레임워크 충돌을 줄이기 위한 환경변수 가드. - TORCH_GPU_ONLY: Torch만 GPU, TF/ORT는 CPU - ORT_GPU: ORT GPU, TF는 CPU - TF_GPU_ONLY: TF만 GPU """ os.environ.setdefault("CUDA_VISIBLE_DEVICES", str(settings.CUDA_DEVICE)) os.environ.setdefault("TF_CPP_MIN_LOG_LEVEL", "2") os.environ.setdefault("TF_FORCE_GPU_ALLOW_GROWTH", "true") if settings.FRAMEWORK_POLICY == "TORCH_GPU_ONLY": os.environ["TF_VISIBLE_DEVICES"] = "" # TF는 GPU 비노출 elif settings.FRAMEWORK_POLICY == "ORT_GPU": os.environ["TF_VISIBLE_DEVICES"] = "" # TF는 CPU 고정 elif settings.FRAMEWORK_POLICY == "TF_GPU_ONLY": # Torch 비활성은 코드단에서 선택적으로 처리 pass logger.info( f"[ENV GUARD] policy={settings.FRAMEWORK_POLICY}, " f"TF_ENABLED={settings.TF_ENABLED}, ORT_ENABLED={settings.ORT_ENABLED}, " f"ORT_PROVIDERS={settings.ORT_PROVIDERS}" ) def framework_smoketest() -> None: """ 선택적 스모크 테스트: 배포 환경 검증용 """ try: import torch # noqa import torch.cuda # noqa logger.info(f"Torch CUDA 사용 가능: {torch.cuda.is_available()}") if torch.cuda.is_available(): logger.info(f"GPU[{settings.CUDA_DEVICE}]: {torch.cuda.get_device_name(settings.CUDA_DEVICE)}") except Exception as e: logger.error(f"Torch 점검 실패: {e}") if settings.TF_ENABLED: try: import tensorflow as tf # noqa try: if settings.FRAMEWORK_POLICY != "TF_GPU_ONLY": tf.config.set_visible_devices([], "GPU") except Exception: pass logger.info(f"TF {tf.__version__}, GPUs: {tf.config.list_physical_devices('GPU')}") except Exception as e: logger.error(f"TF 점검 실패: {e}") if settings.ORT_ENABLED: try: import onnxruntime as ort # noqa logger.info(f"ONNXRuntime {ort.__version__}, Providers: {ort.get_available_providers()}") except Exception as e: logger.error(f"ONNXRuntime 점검 실패: {e}")