diff --git a/.github/labeler.yml b/.github/labeler.yml new file mode 100644 index 0000000..8f69fee --- /dev/null +++ b/.github/labeler.yml @@ -0,0 +1,31 @@ +model: + - changed-files: + - any-glob-to-any-file: ["src/model/**", "models/**", "*.pt", "*.pth"] + +data: + - changed-files: + - any-glob-to-any-file: ["data/**", "datasets/**", "src/data/**"] + +training: + - changed-files: + - any-glob-to-any-file: ["train*.py", "src/train/**", "configs/**"] + +api: + - changed-files: + - any-glob-to-any-file: ["api/**", "routers/**", "app.py"] + +test: + - changed-files: + - any-glob-to-any-file: ["tests/**", "test_*.py"] + +docs: + - changed-files: + - any-glob-to-any-file: ["docs/**", "*.md", "*.rst"] + +ci: + - changed-files: + - any-glob-to-any-file: [".github/**"] + +dependencies: + - changed-files: + - any-glob-to-any-file: ["requirements*.txt", "pyproject.toml", "setup.py"] \ No newline at end of file diff --git a/.github/workflows/issue-auto-label.yml b/.github/workflows/issue-auto-label.yml deleted file mode 100644 index ceec4a2..0000000 --- a/.github/workflows/issue-auto-label.yml +++ /dev/null @@ -1,14 +0,0 @@ -name: Auto Label Issues - -on: - issues: - types: [opened] - -jobs: - label: - runs-on: ubuntu-latest - steps: - - name: Label Bug - uses: actions-ecosystem/action-add-labels@v1 - with: - labels: bug diff --git a/.github/workflows/issue-stale-clean.yml b/.github/workflows/issue-stale-clean.yml deleted file mode 100644 index 7f90ca7..0000000 --- a/.github/workflows/issue-stale-clean.yml +++ /dev/null @@ -1,22 +0,0 @@ -name: Close stale issues - -on: - schedule: - - cron: "0 0 * * *" - -jobs: - stale: - runs-on: ubuntu-latest - steps: - - uses: actions/stale@v9 - with: - days-before-stale: 14 - days-before-close: 7 - stale-issue-label: "stale" - stale-issue-message: > - ⏳ 14일 동안 활동이 없어 자동으로 stale 처리되었습니다. - 추가 정보가 있다면 댓글로 다시 활성화해주세요! - - close-issue-message: > - 🔒 7일 동안 활동이 없어 Issue가 자동으로 닫혔습니다. - 필요하면 언제든 재오픈해주세요 🙂 diff --git a/.github/workflows/pr-ci.yml b/.github/workflows/pr-ci.yml new file mode 100644 index 0000000..4bbaf9f --- /dev/null +++ b/.github/workflows/pr-ci.yml @@ -0,0 +1,29 @@ +name: PR CI + +on: + pull_request: + branches: [main, develop] + +jobs: + lint-and-test: + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: "pip" + + - name: Install dependencies + run: pip install -r requirements.txt + + - name: Lint (ruff) + run: ruff check . --output-format=github + + - name: Format check (ruff) + run: ruff format --check . + + - name: Test (pytest) + run: pytest --tb=short -q \ No newline at end of file diff --git a/.github/workflows/pr-label.yml b/.github/workflows/pr-label.yml new file mode 100644 index 0000000..df3de40 --- /dev/null +++ b/.github/workflows/pr-label.yml @@ -0,0 +1,18 @@ +name: PR Auto Label + +on: + pull_request: + types: [opened, synchronize, reopened] + +jobs: + label: + runs-on: ubuntu-latest + permissions: + contents: read + pull-requests: write + + steps: + - uses: actions/labeler@v5 + with: + repo-token: ${{ secrets.GITHUB_TOKEN }} + sync-labels: true # PR 파일 변경 시 라벨 재동기화 \ No newline at end of file diff --git a/.gitignore b/.gitignore index da944b5..d0e86dc 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,7 @@ npm-debug.log* yarn-debug.log* yarn-error.log* +*.egg-info # [FastAPI / Python - App 폴더 관련] **/__pycache__/ diff --git a/App/celery_app.py b/App/celery_app.py index 027ba15..5503e8b 100644 --- a/App/celery_app.py +++ b/App/celery_app.py @@ -1,18 +1,10 @@ import os -import sys -from pathlib import Path from celery import Celery from dotenv import load_dotenv -current_file = Path(__file__).resolve() -project_root = current_file.parent.parent - -if str(project_root) not in sys.path: - sys.path.insert(0, str(project_root)) - +# ------ .env 파일 가져오기 ------ load_dotenv() -# 기본값 설정 REDIS_HOST = os.getenv("REDIS_HOST", "localhost") REDIS_PORT = os.getenv("REDIS_PORT", 6379) @@ -23,7 +15,11 @@ "deepguard", broker=REDIS_URL, # 메세지를 전달할 브로커 backend=REDIS_URL, # 작업 결과를 저장할 백엔드 - include=["services.inference_svc"] # 비동기 추론 작업 파일 + # 비동기 추론 작업 파일 + include=["services.inference_svc", # process_image_task, process_video_task + "services.explain_svc", # process_explain_image_task + "services.image_svc", # cleanup_anonymous_image, cleanup_image_cam + "services.video_svc"] # cleanup_anonymous_video ) # Celery Instance 상세 설정 diff --git a/App/main.py b/App/main.py index 530812c..14fafc2 100644 --- a/App/main.py +++ b/App/main.py @@ -1,44 +1,35 @@ import os -import sys -from pathlib import Path -import warnings -import logging from dotenv import load_dotenv -# ------ 상위 폴더 경로 설정 -------- -current_file = Path(__file__).resolve() -project_root = current_file.parent.parent # App의 상위 폴더 - -if str(project_root) not in sys.path: - sys.path.insert(0, str(project_root)) - +# ------ .env 파일 가져오기 ------ load_dotenv() # -------- Huggingface_Hub 인증 ---------- if os.getenv("HF_TOKEN"): os.environ["HF_TOKEN"] = os.getenv("HF_TOKEN") +else: + print("[Warning] HF_TOKEN is not set") from fastapi import FastAPI from fastapi.staticfiles import StaticFiles from fastapi.exceptions import RequestValidationError from starlette.exceptions import HTTPException as StarletteHTTPException from fastapi.middleware.cors import CORSMiddleware -from starlette.middleware.sessions import SessionMiddleware -from routes import auth, home, inference, image, video -from utils.common import lifespan -from utils import exc_handler, middleware +# from starlette.middleware.sessions import SessionMiddleware +from routes import auth, home, inference, image, video, explain +from utils import exc_handler, middleware, common -# 가상 인스턴스 생성 -app = FastAPI(lifespan=lifespan) +# 가상 FastAPI 인스턴스 생성 +app = FastAPI(lifespan=common.lifespan) -# StaticFile 등록하기 +# StaticFile 등록 (이미지, 비디오 파일) app.mount("/static", StaticFiles(directory="static"), name="static") -# Cross Origin Resource Sharing +# CORS Setup for cross-origin requests +# FrontEnd: React, BackEnd: FastAPI origins_str = os.getenv("CORS_ALLOWED_ORIGINS") allowed_origins = [origin.strip() for origin in origins_str.split(",")] - app.add_middleware(CORSMiddleware, allow_origins=allowed_origins, allow_methods=["*"], @@ -52,16 +43,17 @@ # app.add_middleware(SessionMiddleware, secret_key=SECRET_KEY, max_age=3600) # 세션 미들웨어 등록 - Redis 이용 -app.add_middleware(middleware.RedisSessionMiddleware, max_age=7200) +app.add_middleware(middleware.RedisSessionMiddleware, max_age=3600) -# 라우터 등록 -app.include_router(auth.router) -app.include_router(home.router) -app.include_router(inference.router) -app.include_router(image.router) -app.include_router(video.router) +# 라우터 등록 (View: Router, Controller: Service) +app.include_router(auth.router) # 로그인, 로그아웃, 회원가입 +app.include_router(home.router) # 세션 유저 정보 가져오기 +app.include_router(inference.router) # 이미지, 비디오 비동기 추론 접수, 추론 값 가져오기 +app.include_router(image.router) # 이미지 삭제 및 가져오기 +app.include_router(video.router) # 비디오 삭제 및 가져오기 +app.include_router(explain.router) # 딥페이크 이미지, 비디오 위조 흔적 표시 -# Custom HTTPException Handler +# 커스텀 예외 처리: HTTPException app.add_exception_handler(StarletteHTTPException, exc_handler.custom_http_exception_handler) -# Custom RequestValidationError Handler +# 커스텀 예외 처리: RequestValidationError app.add_exception_handler(RequestValidationError, exc_handler.validation_exception_handler) \ No newline at end of file diff --git a/App/routes/explain.py b/App/routes/explain.py new file mode 100644 index 0000000..674edcb --- /dev/null +++ b/App/routes/explain.py @@ -0,0 +1,94 @@ +import os +import io +from pydantic import Field +from fastapi import APIRouter, status, Depends +from fastapi.exceptions import HTTPException +from fastapi.responses import JSONResponse +from sqlalchemy import Connection +from db.database import context_get_conn +from schemas.explain_schema import ExplainImageRequest +from services import session_svc, explain_svc, image_svc +from celery_app import celery_app +from celery.result import AsyncResult + +router = APIRouter(prefix="/explain", tags=["explain"]) + +@router.post("/image/{image_id}", status_code=status.HTTP_202_ACCEPTED, + response_class=JSONResponse, summary="딥페이크 이미지 위조 흔적 시각화 비동기 접수") +async def explain_image( + image_id: int, + explain_req: ExplainImageRequest, + conn: Connection = Depends(context_get_conn), + session_user = Depends(session_svc.get_session_user_prt), # 로그인 필수 +): + # 딥페이크 이미지 추론 결과 가져오기 + result = await image_svc.get_image_result(conn, image_id) + + if result.status != "SUCCESS": + raise HTTPException( + status_code = status.HTTP_400_BAD_REQUEST, + detail = "이미지 위조 흔적 분석은 추론이 성공한 이미지만 가능합니다" + ) + + image_path = "." + result.image_loc + + if not os.path.exists(image_path): + raise HTTPException( + status_code = status.HTTP_404_NOT_FOUND, + detail = f"요청하신 이미지 파일을 찾을 수 없습니다. 삭제하였는지 다시 확인해주세요." + ) + + if result.model_type == "pro" and explain_req.aug_smooth: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Pro 모델은 aug_smooth 기능을 지원하지 않습니다", + ) + + # Celery Task 호출(Redis Broker 활용) + task = explain_svc.process_explain_image_task.delay( + user_email = session_user["email"], + result_dict = result.model_dump(mode='json'), + explain_req_dict = explain_req.model_dump()) + return { + "message": "딥페이크 이미지 위조 흔적 시각화 접수 완료. 시각화 분석 시작 ...", + "task_id": task.id, + } + + +@router.get("/image/result/{task_id}", status_code=status.HTTP_200_OK, + response_class=JSONResponse, summary="딥페이크 이미지 위조 흔적 시각화 결과 가져오기") +async def get_explain_image_result( + task_id: str, + session_user = Depends(session_svc.get_session_user_prt), # 로그인 필수 + ): + + # Redis Broker에서 Task ID에 해당하는 비동기 작업 상태 가져오기 + task = AsyncResult(task_id, app=celery_app) + + if task.state in ("PENDING", "STARTED", "RETRY"): + return { + "status": "PENDING", + "message": "딥페이크 이미지 위조 흔적 시각화 분석 중 ..." + } + + if task.state == "FAILURE": + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="딥페이크 이미지 위조 흔적 시각화 중 알 수 없는 오류가 발생하였습니다") + + # Celery Task 결과 가져오기 + result = task.result + + if result["status"] == "FAILED": + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=result["message"]) + + return { + "status": result["status"], + "message": result["message"], + "cam_loc": result["cam_loc"], + } + + + \ No newline at end of file diff --git a/App/routes/inference.py b/App/routes/inference.py index ba856af..9d04a32 100644 --- a/App/routes/inference.py +++ b/App/routes/inference.py @@ -1,22 +1,24 @@ from fastapi import ( - APIRouter, Depends, status, Form, - File, UploadFile -) + APIRouter, UploadFile, status, + Depends, Form, File) +from fastapi.responses import JSONResponse from fastapi.exceptions import HTTPException from services import image_svc, session_svc, inference_svc, video_svc from sqlalchemy import Connection from db.database import context_get_conn -from schemas.video_schema import VideoDetailResponse - +from schemas.video_schema import VideoDetailResponse, VideoDetailData +from schemas.image_schema import ImageData_indi +from typing import Literal router = APIRouter(prefix="/inference", tags=["inference"]) -@router.post("/image", status_code=status.HTTP_202_ACCEPTED, summary="딥페이크 비동기 이미지 추론 접수") +@router.post("/image", status_code=status.HTTP_202_ACCEPTED, + response_class=JSONResponse, summary="딥페이크 비동기 이미지 추론 접수") async def predict_image( imagefile: UploadFile = File(...), # 사용자가 업로드한 이미지 객체 - version_type: str = Form(...), # deepguard1, deepguard2 - model_type: str = Form(...), # fast model, pro moel - domain_type: str = Form(...), # model 학습시 사용한 dataset 종류 + version_type: Literal["v1","v2"] = Form("v2", description="모델 엔진 버전"), + model_type: Literal["fast","pro"] = Form("fast", description="추론 모드 (fast: 속도 우선, pro: 정확도 우선)"), + domain_type: Literal["서양인","동양인"] = Form("서양인", description="학습 데이터셋 도메인"), conn: Connection = Depends(context_get_conn), # 이미지 File 저장 session_user = Depends(session_svc.get_session_user_opt) # Signed Cookie 없을 시 None 반환 ): @@ -43,7 +45,8 @@ async def predict_image( "status": "success", } -@router.get("/image/{image_id}", status_code=status.HTTP_200_OK, summary="딥페이크 비동기 이미지 추론 결과값 가져오기") +@router.get("/image/{image_id}", status_code=status.HTTP_200_OK, + response_model=ImageData_indi, summary="딥페이크 비동기 이미지 추론 결과값 가져오기") async def get_image_result( image_id: int, conn: Connection = Depends(context_get_conn), @@ -62,12 +65,13 @@ async def get_image_result( ) return image_data -@router.post("/video", status_code=status.HTTP_202_ACCEPTED, summary="딥페이크 비동기 비디오 추론 접수") +@router.post("/video", status_code=status.HTTP_202_ACCEPTED, + response_class=JSONResponse, summary="딥페이크 비동기 비디오 추론 접수") async def predict_video( videofile: UploadFile = File(...), # 사용자가 업로드한 비디오 객체 - version_type: str = Form(...), # deepguard1, deepguard2 - model_type: str = Form(...), # fast model, pro moel - domain_type: str = Form(...), # model 학습시 사용한 dataset 종류 + version_type: Literal["v1","v2"] = Form("v2", description="모델 엔진 버전"), + model_type: Literal["fast","pro"] = Form("fast", description="추론 모드 (fast: 속도 우선, pro: 정확도 우선)"), + domain_type: Literal["서양인","동양인"] = Form("서양인", description="학습 데이터셋 도메인"), conn: Connection = Depends(context_get_conn), # 비디오 File 저장 session_user = Depends(session_svc.get_session_user_opt) # Signed Cookie 없을 시 None 반환 ): @@ -94,7 +98,8 @@ async def predict_video( "status": "success", } -@router.get("/video/{video_id}", status_code=status.HTTP_200_OK, summary="딥페이크 비디오 비디오 추론 결과값 가져오기") +@router.get("/video/{video_id}", status_code=status.HTTP_200_OK, + response_model=VideoDetailData, summary="딥페이크 비디오 비디오 추론 결과값 가져오기") async def get_video_result( video_id: int, conn: Connection = Depends(context_get_conn), @@ -117,12 +122,20 @@ async def get_video_result( return video_data -@router.get("/video/{video_id}/detail", status_code=status.HTTP_200_OK, summary="딥페이크 상세 분석 결과값 가져오기") +@router.get("/video/{video_id}/detail", status_code=status.HTTP_200_OK, + response_model=VideoDetailResponse ,summary="딥페이크 상세 분석 결과값 가져오기") async def get_video_detail( video_id: int, conn: Connection = Depends(context_get_conn), - session_user = Depends(session_svc.get_session_user_prt) # 로그인 유저만 가능 + session_user = Depends(session_svc.get_session_user_prt) # 로그인 필수 ): + video_data = await video_svc.get_video_result(conn, video_id) + if video_data.status != "SUCCESS": + raise HTTPException( + status_code = status.HTTP_400_BAD_REQUEST, + detail = f"비디오 상세 분석은 추론이 성공한 비디오만 가능합니다" + ) + meta = await video_svc.get_video_meta_result(conn, video_id) frames = await video_svc.get_video_frame_results(conn, video_id) diff --git a/App/schemas/explain_schema.py b/App/schemas/explain_schema.py new file mode 100644 index 0000000..37a51d8 --- /dev/null +++ b/App/schemas/explain_schema.py @@ -0,0 +1,54 @@ +from typing import Literal +from pydantic import BaseModel, Field, model_validator + +# ── Branch별 허용 기법 ──────────────────────────────────── +# +# [LOW branch] 고해상도 feature map → 국소 위조 흔적 포착에 특화 +# - hirescam : activation * gradient element-wise 곱, 업샘플링 없이 원본 해상도 유지 +# → 픽셀 단위 경계가 가장 선명 (aug_smooth: O, eigen_smooth: X) +# - gradcamelementwise : 각 위치별 gradient 부호 고려 후 ReLU → 노이즈 억제, 국소 영역 정밀 (aug_smooth: O, eigen_smooth: O) +# - layercam : positive gradient로 activation 공간 가중합산 +# → 얕은 레이어에서도 안정적, 텍스처/경계 검출에 강함 (aug_smooth: O, eigen_smooth: O) +# +# [HIGH branch] 저해상도 semantic feature map → 전체 구조 파악에 특화 +# - eigengradcam : activation * gradient 의 첫 번째 주성분 +# → 클래스 판별력 유지하면서 GradCAM보다 부드럽고 EigenCAM보다 정확 (aug_smooth: O, eigen_smooth: X) +# - gradcamplusplus : 2차 gradient 활용 → 다중 위조 영역 동시 포착, 작은 영역에 강함 (aug_smooth: O, eigen_smooth: X) +# - xgradcam : gradient를 normalized activation으로 스케일링 +# → attribution 합이 logit 변화량과 수학적으로 일치, 충실도 높음 (aug_smooth: O, eigen_smooth: O) + +_LOW_ALLOWED = {"hirescam", "gradcamelementwise", "layercam"} +_HIGH_ALLOWED = {"eigengradcam", "gradcamplusplus", "xgradcam"} +_EIGEN_ALLOWED = {"gradcamelementwise", "layercam", "xgradcam"} + + +class ExplainImageRequest(BaseModel): + branch_level: Literal["low","high"] = Field("high", description="브랜치 레벨\nlow: 국소 위조 흔적 포착\nhigh: 전역적 위조 흔적 포착") + explainer_type: str = Field("eigengradcam", description = ("선택 가능한 XAI 기법. low: [hirescam, gradcamelementwise, layercam], ""high: [eigengradcam, gradcamplusplus, xgradcam]")) + display_type: Literal["heatmap", "bbox"] = Field("heatmap", description="시각화 형태. heatmap: 전체 분포, bbox: 위조 의심 영역 사각형") + category: Literal[0, 1] = Field(1, description="판단 클래스 인덱스 (0: Real / 1: Fake)") + overlay_ratio: float = Field(0.5, ge=0.0, le=1.0, description = "Heatmap 투명도 (0: 히트맵만 강조, 1: 원본 이미지 위주)") + threshold: float = Field(0.5, ge=0.5, le=1.0, description="contour/bbox 이진화 임계값 (0.0~1.0)") + aug_smooth: bool = Field(False, description = "TTA(Test Time Augmentation) 적용 여부. 히트맵을 더 객체 중심적으로 정렬") + eigen_smooth: bool = Field(False, description = "PCA 기반 노이즈 제거. 지배적인 패턴만 남김") + + @model_validator(mode="after") + def validate_explainer_for_branch(self) -> "ExplainImageRequest": + branch_allowed = {"low": _LOW_ALLOWED, "high": _HIGH_ALLOWED} + if self.explainer_type not in branch_allowed[self.branch_level]: + raise ValueError( + f"branch_level='{self.branch_level}'에서 허용된 기법: " + f"{sorted(branch_allowed[self.branch_level])} " + f"(입력값: '{self.explainer_type}')" + ) + return self + + @model_validator(mode="after") + def validate_explainer_for_eigen_smooth(self) -> "ExplainImageRequest": + if self.eigen_smooth and self.explainer_type not in _EIGEN_ALLOWED: + raise ValueError( + f"eigen_smooth가 허용된 기법: " + f"{sorted(_EIGEN_ALLOWED)} " + f"(입력값: '{self.explainer_type}')" + ) + return self \ No newline at end of file diff --git a/App/services/explain_svc.py b/App/services/explain_svc.py new file mode 100644 index 0000000..83acc09 --- /dev/null +++ b/App/services/explain_svc.py @@ -0,0 +1,110 @@ +import cv2 +import asyncio +import numpy as np +from functools import partial +from fastapi import status +from fastapi.exceptions import HTTPException +from sqlalchemy import Connection +from services import image_svc, inference_svc +from explainability import ( + CAMExplainer, + HiResCAMExplainer, GradCAMElementWiseExplainer, LayerCAMExplainer, + EigenGradCAMExplainer, GradCAMPlusPlusExplainer, XGradCAMExplainer, +) +from celery_app import celery_app + +# --- CAM Explainer Registry --- +# LOW branch: 국소 위조 흔적 포착 +# HIGH branch: 전역 위조 흔적 포착 +EXPLAINER_REGISTRY = { + "hirescam": HiResCAMExplainer, #LOW branch + "gradcamelementwise": GradCAMElementWiseExplainer, #LOW branch + "layercam": LayerCAMExplainer, #LOW branch + "eigengradcam": EigenGradCAMExplainer, # HIGH branch + "gradcamplusplus": GradCAMPlusPlusExplainer, # HIGH branch + "xgradcam": XGradCAMExplainer, # HIGH branch +} + +_explainer_cache: dict = {} + +# 캐시된 CAMExplainer 객체 반환하거나 새로 생성 +def _get_or_create_explainer(model_name: str, dataset: str, explain_req_dict: dict): + cache_key = (model_name, dataset, explain_req_dict["explainer_type"], explain_req_dict["branch_level"]) + + if cache_key not in _explainer_cache: + _explainer_cache[cache_key] = EXPLAINER_REGISTRY[explain_req_dict["explainer_type"]]( + model_name = model_name, + dataset = dataset, + branch_level = explain_req_dict["branch_level"], + ) + return _explainer_cache[cache_key] + +# 시각화 이미지 생성 (heatmap, contour, bbox 선택) +def _run_visualization(explainer: CAMExplainer, image_path: str, explain_req_dict: dict) -> np.ndarray: + if explain_req_dict["display_type"] == "heatmap": + return explainer.display_heatmap_on_image(image_path, image_weight=explain_req_dict["overlay_ratio"], threshold=explain_req_dict["threshold"], + category=explain_req_dict["category"], aug_smooth=explain_req_dict["aug_smooth"], eigen_smooth=explain_req_dict["eigen_smooth"]) + else: + return explainer.display_bbox_on_image(image_path, threshold=explain_req_dict["threshold"], + category=explain_req_dict["category"], aug_smooth=explain_req_dict["aug_smooth"], eigen_smooth=explain_req_dict["eigen_smooth"]) + +# 딥페이크 이미지 위조 흔적 시각화 처리 +@celery_app.task(name="process_explain_image_task") +def process_explain_image_task(user_email: str, + result_dict: dict, + explain_req_dict: dict): + async def run_explain(): + cam_loc = None + try: + # 추론에 사용된 모델 및 데이터셋 설정 로드 + version_type = result_dict["version_type"] + model_type = result_dict["model_type"] + domain_type = result_dict["domain_type"] + + model_name, dataset = inference_svc.MODEL_CONFIG[version_type][model_type][domain_type] + + explainer = _get_or_create_explainer(model_name, dataset, explain_req_dict) + image_path = "." + result_dict["image_loc"] + + # 시각화 이미지 생성 시작 + try: + image = _run_visualization(explainer, image_path, explain_req_dict) + except Exception: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="딥페이크 이미지 위조 흔적을 생성하는 중 오류가 발생하였습니다" + ) + + # 생성된 시각화 이미지 파일 저장 + try: + cam_loc = await image_svc.upload_image_cam(user_email, result_dict["image_id"], image) + except Exception: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="딥페이크 이미지 위조 흔적 파일을 저장하는 중 오류가 발생했습니다." + ) + + return {"status": "SUCCESS", + "message": "딥페이크 이미지 위조 흔적 시각화가 성공적으로 이루어졌습니다", + "cam_loc": cam_loc} + + except HTTPException as e: + print(e.detail) + return {"status": "FAILED", "message": str(e.detail)} + + except Exception as e: + print(str(e)) + return {"status": "FAILED", "message": str(e)} + + finally: + # 임시 저장된 시각화 이미지 파일 삭제 + if cam_loc: + image_svc.cleanup_image_cam.apply_async(args=[cam_loc], countdown=60) + + # 동기식 Celery 워커 내 비동기 이벤트 루프 구동 + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + return loop.run_until_complete(run_explain()) + finally: + loop.close() diff --git a/App/services/image_svc.py b/App/services/image_svc.py index 9a92d68..f83194e 100644 --- a/App/services/image_svc.py +++ b/App/services/image_svc.py @@ -1,47 +1,44 @@ import os -import asyncio import time +import cv2 +import numpy as np import aiofiles as aio +import asyncio from dotenv import load_dotenv - from fastapi import UploadFile, status from fastapi.exceptions import HTTPException from sqlalchemy import text, Connection -from sqlalchemy.exc import SQLAlchemyError, DBAPIError +from sqlalchemy.exc import SQLAlchemyError from schemas.image_schema import UserHistory, UserHistory_indi, ImageData_indi from db.database import celery_db_conn from celery_app import celery_app load_dotenv() -UPLOAD_DIR = os.getenv("UPLOAD_DIR") +UPLOAD_DIR = os.getenv("UPLOAD_DIR", "./static/uploads") +EXPLAIN_UPLOAD_DIR = os.getenv("EXPLAIN_UPLOAD_DIR", "./static/explain") + +os.makedirs(EXPLAIN_UPLOAD_DIR, exist_ok=True) +os.makedirs(UPLOAD_DIR, exist_ok=True) -# 사용자 업로드 이미지 서버 내 저장 (회원/비회원 공통) +# 사용자 업로드 이미지 파일 서버 내 저장 (회원/비회원 공통) async def upload_image(user_email: str | None, imagefile: UploadFile) -> str: try: - # 1. 사용자별 하위 디렉토리 결정 + # 사용자별 하위 디렉토리 결정 sub_dir = user_email if user_email else "anonymous" user_dir = os.path.join(UPLOAD_DIR, sub_dir) - # 3. 디렉토리 존재 확인 및 생성 - if not os.path.exists(user_dir): - try: - os.makedirs(user_dir, exist_ok=True) - except OSError as e: - print(f"[Storage Error] 디렉토리 생성 실패: {e}") - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="서버 내 저장 공간을 준비하지 못했습니다." - ) - - # 4. 파일명 중복 방지 (파일명_타임스탬프.확장자) + # 디렉토리 존재 확인 및 생성 + os.makedirs(user_dir, exist_ok=True) + + # 파일명 중복 방지 (파일명_타임스탬프.확장자) filename_only, ext = os.path.splitext(imagefile.filename) upload_filename = f"{filename_only}_{int(time.time())}{ext}" upload_image_loc = os.path.join(user_dir, upload_filename) - # 5. 비동기 이미지 저장 (Chunk 단위 읽기) + # 비동기 이미지 저장 (Chunk 단위 읽기) try: async with aio.open(upload_image_loc, "wb") as outfile: while content := await imagefile.read(1024 * 1024): @@ -53,8 +50,6 @@ async def upload_image(user_email: str | None, imagefile: UploadFile) -> str: detail="이미지 파일을 저장하는 중 오류가 발생했습니다." ) - print(f"Upload Succeeded: {upload_image_loc}") - # 6. DB 저장용 경로 반환 return upload_image_loc[1:].replace("\\", "/") @@ -67,6 +62,25 @@ async def upload_image(user_email: str | None, imagefile: UploadFile) -> str: status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="이미지 업로드 과정에서 예상치 못한 오류가 발생했습니다.") +# 딥페이크 이미지 위조 흔적 시각화 파일 서버 내 저장 (회원 전용) +async def upload_image_cam(user_email: str, image_id: int, image: np.ndarray) -> str: + user_dir = os.path.join(EXPLAIN_UPLOAD_DIR, user_email) + os.makedirs(user_dir, exist_ok=True) + + cam_filename = f"{image_id}_{int(time.time())}.png" + cam_loc = os.path.join(user_dir, cam_filename) + + image_bgr = cv2.cvtColor(image, cv2.COLOR_RGB2BGR) + _, buf = cv2.imencode(".png", image_bgr) + + try: + async with aio.open(cam_loc, "wb") as outfile: + await outfile.write(buf.tobytes()) + except Exception as e: + raise e + + return cam_loc[1:].replace("\\", "/") + # 사용자 업로드 이미지 서버 내 삭제 async def delete_image(image_loc: str): try: @@ -75,12 +89,10 @@ async def delete_image(image_loc: str): if os.path.exists(file_path): os.remove(file_path) - print(f"File removed: {file_path}") else: print(f"File not found: {file_path}") - except Exception as e: - print(f"{e}") + except Exception: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="알수없는 이유로 문제가 발생하였습니다." @@ -188,7 +200,23 @@ async def _delete(): finally: loop.close() - +@celery_app.task(name="cleanup_image_cam") +def cleanup_image_cam(cam_loc: str): + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + async def _delete(): + try: + await delete_image(cam_loc) + print(f"[Cleanup] 이미지 시각화 파일 삭제 완료 - cam_loc: {cam_loc}") + except Exception as e: + print(f"[Cleanup] 이미지 시각화 파일 삭제 실패 - cam_loc: {cam_loc}, error: {e}") + + loop.run_until_complete(_delete()) + + finally: + loop.close() + # 이미지 DB 레코드 및 물리 파일 완전 삭제 async def delete_image_db(conn: Connection, image_id: int): try: @@ -200,9 +228,7 @@ async def delete_image_db(conn: Connection, image_id: int): await conn.commit() - except SQLAlchemyError as e: - print(e) await conn.rollback() raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="요청하신 서비스가 잠시 내부적으로 문제가 발생하였습니다.") diff --git a/App/services/inference_svc.py b/App/services/inference_svc.py index 3ca92d6..1a1d440 100644 --- a/App/services/inference_svc.py +++ b/App/services/inference_svc.py @@ -1,17 +1,15 @@ import asyncio -from inference.image_predictor_prt import ImagePredictor -from inference.video_predictor_prt import VideoPredictor -from inference.utils import PredictorError +from typing import Literal from fastapi import status, Depends from fastapi.exceptions import HTTPException from sqlalchemy import text, Connection from sqlalchemy.exc import SQLAlchemyError from services import image_svc, video_svc -from db.database import context_get_conn, celery_db_conn, engine +from db.database import celery_db_conn from celery_app import celery_app - -image_cache = {} -video_cache = {} +from inference.image_predictor_prt import ImagePredictor +from inference.video_predictor_prt import VideoPredictor +from inference.utils import PredictorError # 사용자 모델 설정 변수명 MODEL_CONFIG = { @@ -31,29 +29,46 @@ } } -# 사용자 이미지 딥페이크 여부 판단 로직 -def predict_image(image_loc: str, version_type: str, model_type: str, domain_type: str): - - # image_loc: static 폴더 내 저장된 사용자 이미지 경로 - # version_type: v1, v2 - # model_type: fast, pro - # domain_type: 서양인, 동양인 - - model_name, dataset = MODEL_CONFIG[version_type][model_type][domain_type] +_image_cache: dict = {} +_video_cache: dict = {} - # 캐시 확인 및 모델 초기화 + # 캐시 확인 및 모델 초기화 +def _get_or_create_predictor( + model_name: str, + dataset: str, + predictor_mode: Literal["image", "video"], +) -> ImagePredictor | VideoPredictor: cache_key = (model_name, dataset) - if cache_key not in image_cache: - image_cache[cache_key] = ImagePredictor( + + if predictor_mode == "image": + if cache_key not in _image_cache: + _image_cache[cache_key] = ImagePredictor( + margin_ratio=0.2, + conf_thres=0.5, + model_name=model_name, + dataset=dataset, + ) + return _image_cache[cache_key] + + if cache_key not in _video_cache: + _video_cache[cache_key] = VideoPredictor( margin_ratio=0.2, conf_thres=0.5, model_name=model_name, - dataset=dataset + dataset=dataset, ) - predictor = image_cache[cache_key] + return _video_cache[cache_key] + +# 사용자 이미지 딥페이크 여부 판단 로직 +def predict_image(image_loc: str, version_type: str, model_type: str, domain_type: str): + + model_name, dataset = MODEL_CONFIG[version_type][model_type][domain_type] + + predictor = _get_or_create_predictor(model_name, dataset, "image") + image_path = "." + image_loc try: - analysis = predictor.predict_img("." + image_loc) + analysis = predictor.predict_img(image_path) print(f"딥페이크 확률 값: {analysis['prob']}, 얼굴 신뢰도: {analysis['face_conf']}, 얼굴 비율: {analysis['face_ratio']}, 얼굴 밝기: {analysis['face_brightness']}") @@ -130,26 +145,13 @@ async def run_inference(): # 사용자 비디오 딥페이크 여부 판단 로직 def predict_video(video_loc: str, version_type: str, model_type: str, domain_type: str): - # video_loc: static 폴더 내 저장된 사용자 비디오 경로 - # version_type: v1, v2 - # model_type: fast, pro - # domain_type: 서양인, 동양인 - model_name, dataset = MODEL_CONFIG[version_type][model_type][domain_type] - # 캐시 확인 및 모델 초기화 - cache_key = (model_name, dataset) - if cache_key not in video_cache: - video_cache[cache_key] = VideoPredictor( - margin_ratio=0.2, - conf_thres=0.5, - model_name=model_name, - dataset=dataset - ) - predictor = video_cache[cache_key] + predictor = _get_or_create_predictor(model_name, dataset, "video") + video_path = "." + video_loc try: - analysis = predictor.predict_video("." + video_loc) + analysis = predictor.predict_video(video_path) print(f"딥페이크 확률 값: {analysis['prob']}, 얼굴 신뢰도: {analysis['face_conf']}, 얼굴 비율: {analysis['face_ratio']}, 얼굴 밝기: {analysis['face_brightness']}") @@ -207,7 +209,7 @@ async def run_inference(): if result["analysis"].get("frame_results"): try: async with celery_db_conn() as conn: - await video_svc.save_video_frame_results( + await video_svc.save_video_frame_result( conn, video_id, result["analysis"]["frame_results"] ) except Exception as e: diff --git a/App/services/video_svc.py b/App/services/video_svc.py index cdf5ac4..fd69f20 100644 --- a/App/services/video_svc.py +++ b/App/services/video_svc.py @@ -52,8 +52,6 @@ async def upload_video(user_email: str | None, videofile: UploadFile) -> str: detail="비디오 파일을 저장하는 중 오류가 발생했습니다." ) - print(f"Upload Succeeded: {upload_video_loc}") - # 6. DB 저장용 경로 반환 return upload_video_loc[1:].replace("\\", "/") @@ -73,7 +71,6 @@ async def delete_video(video_loc: str): if os.path.exists(file_path): os.remove(file_path) - print(f"Video file removed: {file_path}") else: print(f"Video file not found: {file_path}") @@ -347,7 +344,7 @@ async def save_video_meta_result(conn: Connection, video_id: int, analysis: dict # 비디오 상세 결과 값 저장하기 # 비디오 프레임 별 딥페이크 점수, 얼굴 신뢰도, 얼굴 비율, 얼굴 조도 반환 -async def save_video_frame_results(conn: Connection, video_id: int, frame_results: list): +async def save_video_frame_result(conn: Connection, video_id: int, frame_results: list): try: query = """ INSERT INTO video_frame_result diff --git a/App/utils/common.py b/App/utils/common.py index c49d71c..a72d30c 100644 --- a/App/utils/common.py +++ b/App/utils/common.py @@ -2,12 +2,20 @@ from db.database import engine from contextlib import asynccontextmanager +# Uvicorn이 앱 시작/종료 시 asynccontextmanager를 호출함 @asynccontextmanager async def lifespan(app: FastAPI): - # FastAPI 인스턴스 기동시 필요한 작업 수행 + + # ── Startup ────────────────────────────────────────── + # Event Loop 위에서 실행, DB 연결 풀 초기화 print("Starting up...") + + # yield 이후 app이 실제 요청을 처리하기 시작함 yield - # FastAPI 인스턴스 종료시 필요한 작업 수행 + # ── Shutdown ───────────────────────────────────────── + # 새 요청은 차단되고, 진행 중인 요청 완료 후 여기로 진입 print("Shutting down...") + + # SQLAlchemy 비동기 엔진의 커넥션 풀 전체 반환 및 종료 await engine.dispose() \ No newline at end of file diff --git a/App/utils/responses.py b/App/utils/responses.py new file mode 100644 index 0000000..4647454 --- /dev/null +++ b/App/utils/responses.py @@ -0,0 +1,4 @@ +from fastapi.responses import StreamingResponse + +class PNGStreamingResponse(StreamingResponse): + media_type = "image/png" \ No newline at end of file diff --git a/Images/deepfake-xai/ms_eff_vit_b0_high_eigengradcam.JPG b/Images/deepfake-xai/ms_eff_vit_b0_high_eigengradcam.JPG new file mode 100644 index 0000000..cee217c Binary files /dev/null and b/Images/deepfake-xai/ms_eff_vit_b0_high_eigengradcam.JPG differ diff --git a/Images/deepfake-xai/ms_eff_vit_b0_high_gradcamplusplus.JPG b/Images/deepfake-xai/ms_eff_vit_b0_high_gradcamplusplus.JPG new file mode 100644 index 0000000..21c3a61 Binary files /dev/null and b/Images/deepfake-xai/ms_eff_vit_b0_high_gradcamplusplus.JPG differ diff --git a/Images/deepfake-xai/ms_eff_vit_b0_high_xgradcam.JPG b/Images/deepfake-xai/ms_eff_vit_b0_high_xgradcam.JPG new file mode 100644 index 0000000..2362109 Binary files /dev/null and b/Images/deepfake-xai/ms_eff_vit_b0_high_xgradcam.JPG differ diff --git a/Images/deepfake-xai/ms_eff_vit_b0_low_gradcamelementwise.JPG b/Images/deepfake-xai/ms_eff_vit_b0_low_gradcamelementwise.JPG new file mode 100644 index 0000000..97608a4 Binary files /dev/null and b/Images/deepfake-xai/ms_eff_vit_b0_low_gradcamelementwise.JPG differ diff --git a/Images/deepfake-xai/ms_eff_vit_b0_low_hirescam.JPG b/Images/deepfake-xai/ms_eff_vit_b0_low_hirescam.JPG new file mode 100644 index 0000000..9a282e7 Binary files /dev/null and b/Images/deepfake-xai/ms_eff_vit_b0_low_hirescam.JPG differ diff --git a/Images/deepfake-xai/ms_eff_vit_b0_low_layercam.JPG b/Images/deepfake-xai/ms_eff_vit_b0_low_layercam.JPG new file mode 100644 index 0000000..983d8f1 Binary files /dev/null and b/Images/deepfake-xai/ms_eff_vit_b0_low_layercam.JPG differ diff --git a/Images/deepfake-xai/ms_eff_vit_b5_high_eigengradcam.JPG b/Images/deepfake-xai/ms_eff_vit_b5_high_eigengradcam.JPG new file mode 100644 index 0000000..186a1bc Binary files /dev/null and b/Images/deepfake-xai/ms_eff_vit_b5_high_eigengradcam.JPG differ diff --git a/Images/deepfake-xai/ms_eff_vit_b5_high_gradcamplusplus.JPG b/Images/deepfake-xai/ms_eff_vit_b5_high_gradcamplusplus.JPG new file mode 100644 index 0000000..14a0b75 Binary files /dev/null and b/Images/deepfake-xai/ms_eff_vit_b5_high_gradcamplusplus.JPG differ diff --git a/Images/deepfake-xai/ms_eff_vit_b5_high_xgradcam.JPG b/Images/deepfake-xai/ms_eff_vit_b5_high_xgradcam.JPG new file mode 100644 index 0000000..1965aef Binary files /dev/null and b/Images/deepfake-xai/ms_eff_vit_b5_high_xgradcam.JPG differ diff --git a/Images/deepfake-xai/ms_eff_vit_b5_low_gradcamelementwise.JPG b/Images/deepfake-xai/ms_eff_vit_b5_low_gradcamelementwise.JPG new file mode 100644 index 0000000..5c878a7 Binary files /dev/null and b/Images/deepfake-xai/ms_eff_vit_b5_low_gradcamelementwise.JPG differ diff --git a/Images/deepfake-xai/ms_eff_vit_b5_low_hirescam.JPG b/Images/deepfake-xai/ms_eff_vit_b5_low_hirescam.JPG new file mode 100644 index 0000000..568d4b5 Binary files /dev/null and b/Images/deepfake-xai/ms_eff_vit_b5_low_hirescam.JPG differ diff --git a/Images/deepfake-xai/ms_eff_vit_b5_low_layercam.JPG b/Images/deepfake-xai/ms_eff_vit_b5_low_layercam.JPG new file mode 100644 index 0000000..2225fd2 Binary files /dev/null and b/Images/deepfake-xai/ms_eff_vit_b5_low_layercam.JPG differ diff --git "a/Images/deepfake-xai/\354\204\234\354\226\221\354\235\270_\352\260\200\354\247\234_1.JPG" "b/Images/deepfake-xai/\354\204\234\354\226\221\354\235\270_\352\260\200\354\247\234_1.JPG" new file mode 100644 index 0000000..85a0e1a Binary files /dev/null and "b/Images/deepfake-xai/\354\204\234\354\226\221\354\235\270_\352\260\200\354\247\234_1.JPG" differ diff --git a/README.md b/README.md index e481c00..af97bc7 100644 --- a/README.md +++ b/README.md @@ -1,14 +1,17 @@ # Deepfakes Detection -
-
-
+
+
+
+
+
@@ -43,6 +46,8 @@
- [📈 Model Evaluation](#-model-evaluation) - Benchmarking results
- [💻 Model Usage](#-model-usage) - How to integrate DeepGuard models into your own Python code or via timm
- [🔮 Predict Image & Video](#-predict-image--video) - Simple Inference examples for detecting deepfakes in image and video
+- [🎨 DeepFake AI Explainability](#-deepfake-ai-explainability) - Visualizing model focus using Grad-CAM and attention maps
+- [📊 Metrics and Evaluation for DeepFake XAI](#-metrics-and-evaluation-for-deepfake-xai) - Quantitative assessment of explanation reliability
- [📬 Authors](#-authors)
- [📝 Reference](#-reference)
- [⚖️ License](#-license)
@@ -303,6 +308,41 @@ print(f"Deepfake Probability: {result:.4f}")
```
+## 🎨 DeepFake AI Explainability
+
+Deepfake detection requires high reliability and interpretability. DeepGuard integrates a robust Explainable AI (XAI) Toolkit to visualize the decision-making process of our hybrid models
+
+⭐ Validated on hybrid CNN-ViT architectures, specifically `MS-EffViT` and `MS-EffGCViT`.
+⭐ Dual-Branch Analysis: `Low-Level Branch`(focus on local forgery region), `High-Level Branch`(focus on global forgery region)
+
+### Deepfake Detection: XAI Methods by Multi-Scale Branch
+
+| Branch | Method | 🎯 What it does |
+| :--- | :--- | :--- |
+|  | **HiResCAM** | Like GradCAM but element-wise multiply the activations with the gradients; provably guaranteed faithfulness for certain models |
+|  | **GradCAMElementWise** | Like GradCAM but element-wise multiply the activations with the gradients then apply a ReLU operation before summing |
+|  | **LayerCAM** | Spatially weight the activations by positive gradients. Works better especially in lower layers |
+| --- | --- | --- | --- |
+|  | **EigenGradCAM** | Like EigenCAM but with class discrimination: First principle component of Activations*Grad. Looks like GradCAM, but cleaner |
+|  | **GradCAM++** | Like GradCAM but uses second order gradients |
+|  | **XGradCAM** | Like GradCAM but scale the gradients by the normalized activations |
+
+### Multi Scale Efficient Vision Transformer
+
+| Model | Branch-Level | Image | HiresCam | GradCamElementwise | LayerCam |
+| :--- | :---: | :---: | :---: | :---: | :---: |
+| **⚡ ms-eff-vit-b0** |  | |
|
|
|
+| **⚡ ms-eff-vit-b5** |  |
|
|
|
|
+
+| Model | Branch-Level | Image | EigenGradCam | GradCamPlusPlus | XGradCam |
+| :--- | :---: | :---: | :---: | :---: | :---: |
+| **⚡ ms-eff-vit-b0** |  |
|
|
|
|
+| **🔥 ms-eff-vit-b5** |  |
|
|
|
|
+
+
+
+## 📊 Metrics and Evaluation for DeepFake XAI
+
## 📬 Authors
@@ -323,6 +363,7 @@ _**This project was developed as a Senior Graduation Project by the Department o
6. [`deepfake-detection-project-v4`](https://github.com/ameencaslam/deepfake-detection-project-v4) - _Multiple Deep Learning Models by Ameen Caslam_
7. [`Awesome-Deepfake-Detection`](https://github.com/Daisy-Zhang/Awesome-Deepfakes-Detection
) - _A curated list of tools, papers and code by Daisy Zhang_
+8. [`Pytorch-Grad-Cam`](https://github.com/jacobgil/pytorch-grad-cam) - _Advanced Visual Explanations for PyTorch Models_
## ⚖️ License
diff --git a/explainability/__init__.py b/explainability/__init__.py
index e69de29..b547123 100644
--- a/explainability/__init__.py
+++ b/explainability/__init__.py
@@ -0,0 +1,13 @@
+from explainability.explainer.base_explainer import BaseExplainer
+from explainability.explainer.cam_explainer import CAMExplainer
+from explainability.explainer.gradient_explainer import (
+ GradCAMExplainer, GradCAMPlusPlusExplainer, GradCAMElementWiseExplainer,
+ XGradCAMExplainer, HiResCAMExplainer
+)
+from explainability.explainer.eigenvalue_explainer import (
+ EigenCAMExplainer, EigenGradCAMExplainer, KPCACAMExplainer
+)
+from explainability.explainer.perturbation_explainer import (
+ ScoreCAMExplainer, FEMExplainer
+)
+from explainability.explainer.misc_explainer import LayerCAMExplainer
\ No newline at end of file
diff --git a/explainability/base_explainer.py b/explainability/base_explainer.py
deleted file mode 100644
index 1a54186..0000000
--- a/explainability/base_explainer.py
+++ /dev/null
@@ -1,113 +0,0 @@
-import cv2
-import timm
-import torch
-import numpy as np
-from pathlib import Path
-from typing import List, Optional, Tuple
-from preprocess.face_detector import FaceDetector2
-from deepguard.data import get_test_transforms
-from explainability.utils.reshape_transforms import reshape_transform_vit, reshape_transform_gcvit
-
-DATASETS = {"celeb_df_v2", "ff++", "kodf"}
-MODELS = {"ms_eff_vit_b0", "ms_eff_vit_b5", "ms_eff_gcvit_b0", "ms_eff_gcvit_b5"}
-
-class BaseExplainer:
- def __init__(
- self,
- model_name: str, # ms_eff_vit_b0 | ms_eff_vit_b5 | ms_eff_gcvit_b0 | ms_eff_gcvit_b5
- dataset: str, # celeb_df_v2 | ff++ | both
- margin_ratio: float = 0.2,
- conf_thres: float = 0.5,
- category: int = 1, # 0: real | 1: fake
- branch_level: str = "both", # low | high | both
- l_stage_idx: int = -1 # gcvit low branch stage index (0~3)
- ):
- self._validate_inputs(model_name, dataset, branch_level)
-
- self.device = "cuda:0" if torch.cuda.is_available() else 'cpu'
- self.margin_ratio = margin_ratio
- self.category = category
- self.model_name = model_name
- self.branch_level = branch_level
- self.l_stage_idx = l_stage_idx
-
- # Parse Model Metadata
- self.model_variant = model_name.split("_")[-1] # b0, b5
- self.transformer_type = model_name.split("_")[-2] # vit, gcvit
-
- # Model & Tools setup
- self.model = timm.create_model(model_name, pretrained=True, dataset=dataset)
- self.face_detector = FaceDetector2(conf_thres)
- self.img_size = [224,224] if self.model_variant == "b0" else [384,384]
- self.transforms = get_test_transforms(img_size=self.img_size)
-
- self.model.to(self.device).eval()
-
- # Explainability setup
- self.reshape_fn = self._get_reshape_fn()
- self.target_layers = self._resolve_target_layers()
-
- def _validate_inputs(self, model_name: str, dataset: str, branch_level: str):
- if model_name not in MODELS:
- raise ValueError(f"Unsupported model: '{model_name}'. Must be one of {MODELS}")
- if dataset not in DATASETS:
- raise ValueError(f"Invalid dataset: '{dataset}'. Choose from {DATASETS}")
- if branch_level not in ("low", "high", "both"):
- raise ValueError(f"Invalid branch_level: '{branch_level}'. Expected 'low', 'high', or 'both'.")
-
- def _get_reshape_fn(self):
- return reshape_transform_vit if self.transformer_type == "vit" else reshape_transform_gcvit
-
- def _resolve_target_layers(self) -> list[torch.nn.Module]:
- layers = []
-
- if self.transformer_type == "gcvit":
- n_stages = len(self.model.l_gcvit.stages)
- idx = self.l_stage_idx if self.l_stage_idx >= 0 else n_stages + self.l_stage_idx
-
- if not (0 <= idx < n_stages):
- raise IndexError(f"l_stage_idx {self.l_stage_idx} out of range (n={n_stages})")
- if self.branch_level in ("low", "both"):
- layers.append(self.model.l_gcvit.stages[idx].blocks[-1].norm2)
- if self.branch_level in ("high", "both"):
- layers.append(self.model.h_gcvit.stages[0].blocks[-1].norm2)
-
- else: # vit
- if self.branch_level in ("low", "both"):
- layers.append(self.model.l_vit.blocks[-1].norm1)
- if self.branch_level in ("high", "both"):
- layers.append(self.model.h_vit.blocks[-1].norm1)
-
- return layers
-
- def _get_face_bbox(self, img: np.ndarray) -> List[float]:
- result = self.face_detector.detect_batch([img], [1.0])
- return result[0]
-
- def _crop_face(self, img: np.ndarray, bbox: List[float]) -> np.ndarray:
- xmin, ymin, xmax, ymax = bbox
- h, w = img.shape[:2]
- pw = int((xmax - xmin) * self.margin_ratio)
- ph = int((ymax - ymin) * self.margin_ratio)
- return img[
- max(int(ymin - ph), 0) : min(int(ymax + ph), h),
- max(int(xmin - pw), 0) : min(int(xmax + pw), w),
- ]
-
- def _preprocess_img(self, img_path: str) -> np.ndarray:
- img = cv2.imread(img_path)
- img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
- detect_result = self._get_face_bbox(img)
- bbox = detect_result[:4]
- cropped = self._crop_face(img, bbox)
- return cropped
-
- def _get_transform(self, img_path: str):
- face = self._preprocess_img(img_path)
- tensor = self.transforms(image=face)['image'].unsqueeze(0) # (1,3,H,W)
- return face, tensor
-
- def __repr__(self):
- return (f"{self.__class__.__name__}("
- f"model={self.model_name}, branch={self.branch}, "
- f"device={self.device})")
\ No newline at end of file
diff --git a/explainability/cam_evaluator.py b/explainability/cam_evaluator.py
new file mode 100644
index 0000000..d215def
--- /dev/null
+++ b/explainability/cam_evaluator.py
@@ -0,0 +1,48 @@
+import numpy as np
+import torch
+from typing import List, Dict, Literal
+from explainability.utils.model_targets import BinaryClassifierOutputSigmoidTarget
+from explainability.explainer.cam_explainer import CAMExplainer
+from explainability.metrics.road import (
+ ROADLeastRelevantFirstAverage, ROADMostRelevantFirstAverage,
+ ROADCombined
+)
+
+METRIC_REGISTRY = {
+ "MORF": ROADMostRelevantFirstAverage,
+ "LERF": ROADLeastRelevantFirstAverage,
+ "Combined": ROADCombined,
+}
+
+class CAMEvaluator:
+ def __init__(self,
+ cam_explainer: CAMExplainer,
+ cam_metric: Literal["Combined", "MORF", "LERF"] = "Combined", # Combined, MORF, LERF
+ percentiles: List[int] = [10,20,30,40,50,60,70,80,90],
+ seed: int = 2026
+ ):
+ self.seed = seed
+ self.cam_explainer = cam_explainer
+ self.cam_metric_name = cam_metric
+ self.percentiles = percentiles
+ self.cam_metric = METRIC_REGISTRY[cam_metric](percentiles)
+
+ def evaluate(self, img_path: str) -> float:
+ torch.manual_seed(self.seed)
+ np.random.seed(self.seed)
+
+ grayscale_cam, tensor, _ = self.cam_explainer._build_grayscale_cam(img_path)
+ targets = [BinaryClassifierOutputSigmoidTarget()]
+
+ return self.cam_metric(
+ input_tensor = tensor, # (1,C,H,W)
+ cams = grayscale_cam[np.newaxis, ...], #(1,H,W)
+ targets = targets,
+ model = self.cam_explainer.model,
+ )[0]
+
+ def __repr__(self):
+ return (
+ f"CAMEvaluator(explainer={self.cam_explainer.__class__.__name__}, "
+ f"metric={self.cam_metric_name}, percentiles={self.percentiles})"
+ )
\ No newline at end of file
diff --git a/explainability/cam_explainer.py b/explainability/cam_explainer.py
deleted file mode 100644
index aa5fa96..0000000
--- a/explainability/cam_explainer.py
+++ /dev/null
@@ -1,57 +0,0 @@
-import torch
-import cv2
-import numpy as np
-from abc import ABC, abstractmethod
-from explainability.utils.model_targets import BinaryClassifierOutputTarget
-from explainability.utils.image import show_cam_on_image, deprocess_image
-from .base_explainer import BaseExplainer
-class CAMExplainer(BaseExplainer, ABC):
- def __init__(self,
- aug_smooth: bool = False,
- eigen_smooth: bool = False,
- colormap: int = cv2.COLORMAP_JET,
- image_weight: float = 0.5,
- **kwargs):
- super().__init__(**kwargs)
- self.aug_smooth = aug_smooth # aug_smooth has the effect of beter centering the CAM around the objects
- self.eigen_smooth = eigen_smooth # # eigen_smooth has the effect of removing a lot of noise
- self.colormap = colormap
- self.image_weight = image_weight
- self.cam = self._build_cam()
-
- @abstractmethod
- def _build_cam(self):
- ...
-
- def display_heatmap(self, img_path):
- face, tensor = self._get_transform(img_path)
- tensor = tensor.to(self.device) # (1,3,H,W)
- targets = [BinaryClassifierOutputTarget(self.category)]
-
- grayscale_cam = self.cam(
- input_tensor = tensor, # (1,3,H,W)
- targets = targets,
- aug_smooth = self.aug_smooth,
- eigen_smooth = self.eigen_smooth
- ) # (1,H,W)
-
- return grayscale_cam[0], tensor
-
- def display_cam_on_image(self, img_path):
- grayscale_cam, tensor = self.display_heatmap(img_path)
-
- heatmap = show_cam_on_image(
- deprocess_image(tensor), # (H,W,3), range(0~1), np.ndarray
- grayscale_cam, # (H,W), range(0~1), np.ndarray
- use_rgb=True,
- colormap = self.colormap
- image_weight=self.image_weight)
-
- return heatmap # (H,W,3), range(0~255), np.uint8
-
- def display_inverse_cam(self, img_path):
- grayscale_cam, _ = self.display_heatmap(img_path)
-
- return 1 - grayscale_cam # (H,W), range(0~1), np.float32
-
-
\ No newline at end of file
diff --git a/explainability/explainer/base_explainer.py b/explainability/explainer/base_explainer.py
new file mode 100644
index 0000000..dc4fc62
--- /dev/null
+++ b/explainability/explainer/base_explainer.py
@@ -0,0 +1,194 @@
+import cv2
+import timm
+import torch
+import numpy as np
+from pathlib import Path
+from typing import List, Optional, Tuple, Literal, Callable
+from preprocess.face_detector import FaceDetector2
+from deepguard.data import get_test_transforms
+from explainability.utils.reshape_transforms import reshape_transform_vit, reshape_transform_gcvit
+
+class BaseExplainer:
+ def __init__(
+ self,
+ model_name: Literal["ms_eff_vit_b0", "ms_eff_vit_b5", "ms_eff_gcvit_b0", "ms_eff_gcvit_b5"],
+ dataset: Literal["celeb_df_v2", "ff++", "kodf"],
+ margin_ratio: float = 0.2,
+ conf_thres: float = 0.5,
+ branch_level: Literal["low", "high"] = "high",
+ l_stage_idx: int = -1, # gcvit low branch stage index (0~3)
+ block_idx: int = -1, # vit, gcvit block index
+ ):
+
+ self.device = "cuda:0" if torch.cuda.is_available() else 'cpu'
+ self.margin_ratio = margin_ratio
+ self.conf_thres = conf_thres
+ self.model_name = model_name
+ self.dataset = dataset
+ self.branch_level = branch_level
+ self.l_stage_idx = l_stage_idx
+ self.block_idx = block_idx
+
+ # Parse Model Metadata
+ self.model_variant = model_name.split("_")[-1] # b0, b5
+ self.transformer_type = model_name.split("_")[-2] # vit, gcvit
+
+ # Model & Tools setup
+ self.model = timm.create_model(model_name, pretrained=True, dataset=dataset)
+ self.face_detector = FaceDetector2(conf_thres)
+ self.img_size = [224,224] if self.model_variant == "b0" else [384,384]
+ self.transforms = get_test_transforms(img_size=self.img_size)
+
+ self.model.to(self.device).eval()
+
+ # Explainability setup
+ self.reshape_fn = self._get_reshape_fn()
+ self.target_layers = self._resolve_target_layers()
+
+ def _get_reshape_fn(self) -> Callable[[torch.Tensor], torch.Tensor]:
+ """모델 종류에 맞는 Feature map reshape 함수를 반환한다.
+
+ Returns:
+ Callable: 3D/4D 텐서를 GradCAM 연산에 적합한 형태로 변환하는 함수.
+ """
+ return reshape_transform_vit if self.transformer_type == "vit" else reshape_transform_gcvit
+
+ def _resolve_target_layers(self) -> list[torch.nn.Module]:
+ """시각화(XAI)의 대상이 될 Target Layer들을 선택한다.
+
+ Returns:
+ list[torch.nn.Module]: 분석 대상 정규화(Normalization) 레이어 리스트.
+
+ Raises:
+ ValueError: l_stage_idx나 block_idx가 모델 범위를 벗어날 경우 발생.
+ """
+ layers = []
+
+ if self.transformer_type == "gcvit":
+ n_stages = len(self.model.l_gcvit.stages)
+ stage_idx = self.l_stage_idx if self.l_stage_idx >= 0 else n_stages + self.l_stage_idx
+
+ if not (0 <= stage_idx < n_stages):
+ raise ValueError(f"l_stage_idx={self.l_stage_idx} 범위 초과 (stage 수: {n_stages})")
+
+ if self.branch_level in ("low", "both"):
+ n_blocks = len(self.model.l_gcvit.stages[stage_idx].blocks)
+ blk = self._resolve_block_idx(self.block_idx, n_blocks, f"l_gcvit stage[{stage_idx}]")
+ layers.append(self.model.l_gcvit.stages[stage_idx].blocks[blk].norm2)
+
+ if self.branch_level in ("high", "both"):
+ n_blocks = len(self.model.h_gcvit.stages[0].blocks)
+ blk = self._resolve_block_idx(self.block_idx, n_blocks, "h_gcvit stage[0]")
+ layers.append(self.model.h_gcvit.stages[0].blocks[blk].norm2)
+
+ else: # vit
+ if self.branch_level in ("low", "both"):
+ n_blocks = len(self.model.l_vit.blocks)
+ blk = self._resolve_block_idx(self.block_idx, n_blocks, "l_vit")
+ layers.append(self.model.l_vit.blocks[blk].norm1)
+
+ if self.branch_level in ("high", "both"):
+ n_blocks = len(self.model.h_vit.blocks)
+ blk = self._resolve_block_idx(self.block_idx, n_blocks, "h_vit")
+ layers.append(self.model.h_vit.blocks[blk].norm1)
+
+ return layers
+
+ @staticmethod
+ def _resolve_block_idx(block_idx: int, n_blocks: int, name: str) -> int:
+ """음수 인덱싱을 지원하며, 해당 블록의 유효한 인덱스를 계산 및 검증한다.
+
+ Args:
+ block_idx (int): 접근하려는 블록 인덱스 (음수 가능).
+ n_blocks (int): 해당 스테이지/모델의 총 블록 개수.
+ name (str): 에러 메시지 출력용 모듈 이름.
+
+ Returns:
+ int: 정제된 양수 형태의 블록 인덱스.
+
+ Raises:
+ ValueError: 인덱스가 유효 범위를 벗어난 경우 발생.
+ """
+ idx = block_idx if block_idx >= 0 else n_blocks + block_idx
+ if not (0 <= idx < n_blocks):
+ raise ValueError(f"block_idx={block_idx} 범위 초과 ({name} block 수: {n_blocks})")
+ return idx
+
+ def _get_face_bbox(self, img: np.ndarray) -> List[float]:
+ """얼굴 탐지기(YOLO)를 통해 이미지에서 얼굴 바운딩 박스를 좌표를 추출한다.
+
+ Args:
+ img (np.ndarray): HWC 형태의 RGB 이미지 데이터.
+
+ Returns:
+ list[float]: [xmin, ymin, xmax, ymax, confidence] 형태의 좌표 리스트.
+ """
+ result = self.face_detector.detect_batch([img], [1.0])
+ return result[0]
+
+ def _crop_face(self, img: np.ndarray, bbox: List[float]) -> np.ndarray:
+ """마진 비율을 고려하여 바운딩 박스 기준으로 얼굴 영역을 크롭한다.
+
+ Args:
+ img (np.ndarray): 전체 원본 이미지.
+ bbox (list[float]): 탐지된 [xmin, ymin, xmax, ymax] 얼굴 좌표.
+
+ Returns:
+ np.ndarray: 크롭된 얼굴 이미지 영역.
+ """
+ xmin, ymin, xmax, ymax = bbox
+ h, w = img.shape[:2]
+ pw = int((xmax - xmin) * self.margin_ratio)
+ ph = int((ymax - ymin) * self.margin_ratio)
+ return img[
+ max(int(ymin - ph), 0) : min(int(ymax + ph), h),
+ max(int(xmin - pw), 0) : min(int(xmax + pw), w),
+ ]
+
+ def _preprocess_img(self, img_path: str) -> np.ndarray:
+ """이미지 경로를 읽어 얼굴을 탐지하고 크롭하는 전과정을 수행한다.
+
+ Args:
+ img_path (str): 입력 이미지 파일 경로.
+
+ Returns:
+ np.ndarray: 전처리가 완료된 크롭된 얼굴 이미지 (RGB).
+ """
+ img = cv2.imread(img_path)
+ img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
+ detect_result = self._get_face_bbox(img)
+ bbox = detect_result[:4]
+ cropped = self._crop_face(img, bbox)
+ return cropped
+
+ def _get_transform(self, img_path: str) -> tuple[np.ndarray, torch.Tensor]:
+ """이미지 경로를 받아 크롭 이미지와 모델 입력용 텐서를 동시에 생성한다.
+
+ Args:
+ img_path (str): 입력 이미지 파일 경로.
+
+ Returns:
+ tuple[np.ndarray, torch.Tensor]: (크롭된 얼굴 이미지, (1, 3, H, W) 형태의 배치 텐서).
+ """
+ face = self._preprocess_img(img_path)
+ tensor = self.transforms(image=face)['image'].unsqueeze(0) # (1,3,H,W)
+ return face, tensor
+
+ def _get_transform_from_array(self, face: np.ndarray) -> tuple[np.ndarray, torch.Tensor]:
+ """이미 크롭된 얼굴 배열을 받아 모델 입력용 텐서로 변환한다.
+
+ Args:
+ face (np.ndarray): 이미 잘려진 얼굴 이미지 배열 (RGB).
+
+ Returns:
+ tuple[np.ndarray, torch.Tensor]: (원본 얼굴 이미지, (1, 3, H, W) 형태의 배치 텐서).
+ """
+ tensor = self.transforms(image=face)['image'].unsqueeze(0) # (1,3,H,W)
+ return face, tensor
+
+ def __repr__(self):
+ return (f"{self.__class__.__name__}("
+ f"model={self.model_name}, dataset={self.dataset}, "
+ f"margin_ratio={self.margin_ratio}, conf_thres={self.conf_thres}, "
+ f"branch_level={self.branch_level}, l_stage_idx={self.l_stage_idx}, block_idx={self.block_idx},"
+ f"device={self.device})")
\ No newline at end of file
diff --git a/explainability/explainer/cam_explainer.py b/explainability/explainer/cam_explainer.py
new file mode 100644
index 0000000..f630055
--- /dev/null
+++ b/explainability/explainer/cam_explainer.py
@@ -0,0 +1,334 @@
+from typing import Union, Literal
+import torch
+import cv2
+import numpy as np
+from abc import ABC, abstractmethod
+from explainability.utils.model_targets import BinaryClassifierOutputTarget
+from explainability.utils.image import show_cam_on_image, deprocess_image, remove_padding_and_resize
+from explainability.explainer.base_explainer import BaseExplainer
+
+
+def _draw_label(img: np.ndarray, text: str, x: int, y: int, prob_ratio: float = 1.0):
+ """Confidence label with a clean pill-style background.
+
+ prob_ratio: 0~1, controls bg color (green→red)
+ """
+ font = cv2.FONT_HERSHEY_SIMPLEX
+ font_scale = 0.52
+ thickness = 1
+ pad = 4
+
+ (tw, th), baseline = cv2.getTextSize(text, font, font_scale, thickness)
+
+ rx1, ry1 = x - pad, y - th - pad
+ rx2, ry2 = x + tw + pad, y + baseline + pad
+
+ r = int(min(255, 2 * prob_ratio * 255))
+ g = int(min(255, 2 * (1 - prob_ratio) * 255))
+ bg_color = (r, g, 30) # RGB: green → red
+
+ overlay = img.copy()
+ cv2.rectangle(overlay, (rx1, ry1), (rx2, ry2), bg_color, -1)
+ cv2.addWeighted(overlay, 0.75, img, 0.25, 0, img)
+ cv2.putText(img, text, (x, y), font, font_scale, (255, 255, 255), thickness, cv2.LINE_AA)
+
+
+class CAMExplainer(BaseExplainer, ABC):
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs)
+ self.cam = self._build_cam()
+
+ @abstractmethod
+ def _build_cam(self) -> any:
+ """하위 클래스에서 각 알고리즘(GradCAM, HiResCAM 등)에 맞는 CAM 객체를 생성한다.
+
+ Returns:
+ Any: pytorch-grad-cam 패키지의 CAM 기반 인스턴스.
+ """
+ pass
+
+ def _build_grayscale_cam(self, img_path: str,
+ category: int = 1,
+ aug_smooth: bool = False,
+ eigen_smooth: bool = False) -> tuple[np.ndarray, torch.Tensor, np.ndarray]:
+ """이미지 경로를 받아 전처리를 수행하고 raw 흑백(Grayscale) CAM을 생성한다.
+
+ Args:
+ img_path (str): 입력 이미지 파일 경로.
+ category (int, optional): 타겟 클래스 인덱스 (위조: 1, 정상: 0). Defaults to 1.
+ aug_smooth (bool, optional): Test-time augmentation 가동 여부. Defaults to False.
+ eigen_smooth (bool, optional): PCA 기반 노이즈 제거 가동 여부. Defaults to False.
+
+ Returns:
+ tuple[np.ndarray, torch.Tensor, np.ndarray]:
+ - grayscale_cam: (H, W) 형태의 0~1 사이로 정규화된 CAM 맵.
+ - tensor: (1, 3, H_model, W_model) 형태의 모델 입력용 GPU 텐서.
+ - face: 크롭된 원본 얼굴 이미지 배열 (RGB).
+ """
+
+ face, tensor = self._get_transform(img_path)
+ tensor = tensor.to(self.device)
+ targets = [BinaryClassifierOutputTarget(category)]
+
+ grayscale_cam = self.cam(
+ input_tensor=tensor,
+ targets=targets,
+ aug_smooth=aug_smooth,
+ eigen_smooth=eigen_smooth,
+ )
+ return grayscale_cam[0], tensor, face
+
+ def _build_grayscale_cam_from_array(self,
+ face: np.ndarray,
+ category: int = 1,
+ aug_smooth: bool = False,
+ eigen_smooth: bool = False) -> tuple[np.ndarray, torch.Tensor, np.ndarray]:
+ """이미 크롭된 얼굴 이미지 배열로부터 raw 흑백(Grayscale) CAM을 생성한다.
+
+ Args:
+ face (np.ndarray): 전처리(크롭)가 완료된 얼굴 이미지 배열 (RGB).
+ category (int, optional): 타겟 클래스 인덱스. Defaults to 1.
+ aug_smooth (bool, optional): Test-time augmentation 가동 여부. Defaults to False.
+ eigen_smooth (bool, optional): PCA 기반 노이즈 제거 가동 여부. Defaults to False.
+
+ Returns:
+ tuple[np.ndarray, torch.Tensor, np.ndarray]:
+ - grayscale_cam: (H, W) 형태의 0~1 사이로 정규화된 CAM 맵.
+ - tensor: (1, 3, H_model, W_model) 형태의 모델 입력용 GPU 텐서.
+ - face: 입력받은 원본 얼굴 이미지 배열.
+ """
+ face, tensor = self._get_transform_from_array(face)
+ tensor = tensor.to(self.device)
+ targets = [BinaryClassifierOutputTarget(category)]
+
+ grayscale_cam = self.cam(
+ input_tensor=tensor,
+ targets=targets,
+ aug_smooth=aug_smooth,
+ eigen_smooth=eigen_smooth,
+ )
+ return grayscale_cam[0], tensor, face
+
+ def _prepare_cam(self, img_path: str,
+ category: int = 1,
+ aug_smooth: bool = False,
+ eigen_smooth: bool = False) -> tuple[np.ndarray, np.ndarray]:
+ """흑백 CAM을 생성하고, 모델 입력 시 발생한 패딩 제거 및 원래 얼굴 해상도로 복원한다.
+
+ Args:
+ img_path (str): 입력 이미지 파일 경로.
+ category (int, optional): 타겟 클래스 인덱스. Defaults to 1.
+ aug_smooth (bool, optional): Test-time augmentation 가동 여부. Defaults to False.
+ eigen_smooth (bool, optional): PCA 기반 노이즈 제거 가동 여부. Defaults to False.
+
+ Returns:
+ tuple[np.ndarray, np.ndarray]:
+ - cam_recovered: 원본 얼굴 크기로 복원된 CAM 맵 (H_face, W_face).
+ - face: 크롭된 원본 얼굴 이미지 배열 (RGB).
+ """
+ grayscale_cam, tensor, face = self._build_grayscale_cam(img_path, category, aug_smooth, eigen_smooth)
+ cam_recovered = remove_padding_and_resize(grayscale_cam, face.shape, tensor.shape)
+ return cam_recovered, face
+
+ def _prepare_cam_from_array(self, face: np.ndarray,
+ category: int = 1,
+ aug_smooth: bool = False,
+ eigen_smooth: bool = False) -> tuple[np.ndarray, np.ndarray]:
+ """얼굴 배열로부터 흑백 CAM을 생성하고 원래 얼굴 해상도로 복원한다.
+
+ Args:
+ face (np.ndarray): 전처리(크롭)가 완료된 얼굴 이미지 배열 (RGB).
+ category (int, optional): 타겟 클래스 인덱스. Defaults to 1.
+ aug_smooth (bool, optional): Test-time augmentation 가동 여부. Defaults to False.
+ eigen_smooth (bool, optional): PCA 기반 노이즈 제거 가동 여부. Defaults to False.
+
+ Returns:
+ tuple[np.ndarray, np.ndarray]:
+ - cam_recovered: 원본 얼굴 크기로 복원된 CAM 맵 (H_face, W_face).
+ - face: 입력받은 원본 얼굴 이미지 배열.
+ """
+ grayscale_cam, tensor, face = self._build_grayscale_cam_from_array(face, category, aug_smooth, eigen_smooth)
+ cam_recovered = remove_padding_and_resize(grayscale_cam, face.shape, tensor.shape)
+ return cam_recovered, face
+
+ def _get_binary_mask(self, cam: np.ndarray, threshold: Union[float, Literal["auto"]]) -> np.ndarray:
+ """입력된 CAM 맵을 기반으로 위조 유력 부위를 이진화(Binary Mask)한다.
+
+ Args:
+ cam (np.ndarray): 0~1 사이 값을 가진 복원된 CAM 맵.
+ threshold (Union[float, 'auto']):
+ - float (0.0~1.0): 상위 백분위수 기준으로 임계값 설정 (예: 0.7 이면 상위 30% 영역 선택).
+ - 'auto': Otsu 알고리즘을 사용해 최적의 임계값을 자동으로 계산.
+
+ Returns:
+ np.ndarray: 0 또는 255 값을 가지는 이진화 마스크 맵 (np.uint8).
+ """
+ if threshold != 'auto':
+ t = np.percentile(cam, threshold * 100)
+ binary_mask = (cam > t).astype(np.uint8) * 255
+ else:
+ _, binary_mask = cv2.threshold(
+ np.uint8(cam * 255), 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU
+ )
+ return binary_mask
+
+ def display_heatmap_on_image(self, img_path: str, **kwargs) -> np.ndarray:
+ """이미지 경로를 입력받아 원본 얼굴 위에 위조 흔적 히트맵을 중첩한 이미지를 생성한다.
+
+ Args:
+ img_path (str): 입력 이미지 파일 경로.
+ **kwargs:
+ category (int): 타겟 클래스 인덱스 (기본값: 1).
+ aug_smooth (bool): Test-time augmentation 가동 여부 (기본값: False).
+ eigen_smooth (bool): PCA 기반 노이즈 제거 가동 여부 (기본값: False).
+ threshold (float | str): 이진화 임계값 (기본값: "auto").
+ colormap (int): OpenCV 컬러맵 상수 (기본값: cv2.COLORMAP_JET).
+ image_weight (float): 원본 이미지 불투명도 가중치 (기본값: 0.5).
+
+ Returns:
+ np.ndarray: 히트맵이 합성된 얼굴 이미지 (RGB, np.uint8).
+ """
+ cam_recovered, face = self._prepare_cam(
+ img_path,
+ category = kwargs.get("category", 1),
+ aug_smooth = kwargs.get("aug_smooth", False),
+ eigen_smooth = kwargs.get("eigen_smooth", False))
+
+ binary_mask = self._get_binary_mask(cam_recovered, kwargs.get("threshold", "auto"))
+ cam_recovered = np.where(binary_mask > 0, cam_recovered, 0.0)
+
+ return show_cam_on_image(
+ np.float32(face) / 255.0,
+ cam_recovered,
+ use_rgb = True,
+ colormap = kwargs.get("colormap", cv2.COLORMAP_JET),
+ image_weight = kwargs.get("image_weight", 0.5),
+ )
+
+ def display_heatmap_from_array(self, face: np.ndarray, **kwargs) -> np.ndarray:
+ """얼굴 배열을 직접 입력받아 원본 얼굴 위에 위조 흔적 히트맵을 중첩한 이미지를 생성한다.
+
+ Args:
+ face (np.ndarray): 전처리(크롭)가 완료된 얼굴 이미지 배열 (RGB).
+ **kwargs:
+ category (int): 타겟 클래스 인덱스 (기본값: 1).
+ aug_smooth (bool): Test-time augmentation 가동 여부 (기본값: False).
+ eigen_smooth (bool): PCA 기반 노이즈 제거 가동 여부 (기본값: False).
+ threshold (float | str): 이진화 임계값 (기본값: "auto").
+ colormap (int): OpenCV 컬러맵 상수 (기본값: cv2.COLORMAP_JET).
+ image_weight (float): 원본 이미지 불투명도 가중치 (기본값: 0.5).
+
+ Returns:
+ np.ndarray: 히트맵이 합성된 얼굴 이미지 (RGB, np.uint8).
+ """
+
+ cam_recovered, face = self._prepare_cam_from_array(
+ face,
+ category = kwargs.get("category", 1),
+ aug_smooth = kwargs.get("aug_smooth", False),
+ eigen_smooth = kwargs.get("eigen_smooth", False))
+
+ binary_mask = self._get_binary_mask(cam_recovered, kwargs.get("threshold", "auto"))
+ cam_recovered = np.where(binary_mask > 0, cam_recovered, 0.0)
+
+ return show_cam_on_image(
+ np.float32(face) / 255.0,
+ cam_recovered,
+ use_rgb = True,
+ colormap = kwargs.get("colormap", cv2.COLORMAP_JET),
+ image_weight = kwargs.get("image_weight", 0.5),
+ )
+
+ def display_bbox_on_image(self, img_path: str, **kwargs) -> np.ndarray:
+ """이미지 경로를 입력받아 위조 의심 부위를 찾아 사각형 바운딩 박스와 확률을 그린다.
+
+ Args:
+ img_path (str): 입력 이미지 파일 경로.
+ **kwargs:
+ category (int): 타겟 클래스 인덱스 (기본값: 1).
+ aug_smooth (bool): Test-time augmentation 가동 여부 (기본값: False).
+ eigen_smooth (bool): PCA 기반 노이즈 제거 가동 여부 (기본값: False).
+ threshold (float | str): 이진화 임계값 (기본값: "auto").
+ thickness (int): 사각형 선 굵기 (기본값: 2).
+
+ Returns:
+ np.ndarray: 바운딩 박스와 어노테이션이 추가된 원본 얼굴 이미지 (RGB).
+ """
+ cam_recovered, face = self._prepare_cam(
+ img_path,
+ category = kwargs.get("category", 1),
+ aug_smooth = kwargs.get("aug_smooth", False),
+ eigen_smooth = kwargs.get("eigen_smooth", False)
+ )
+
+ binary_mask = self._get_binary_mask(cam_recovered, kwargs.get("threshold", "auto"))
+
+ contours, _ = cv2.findContours(binary_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+
+ annotated_img = face.copy()
+ for c in contours:
+ x, y, w, h = cv2.boundingRect(c)
+ roi_prob = float(np.max(cam_recovered[y:y + h, x:x + w]))
+
+ r = int(min(255, 2 * roi_prob * 255))
+ g = int(min(255, 2 * (1 - roi_prob) * 255))
+ box_color = (r, g, 30) # RGB
+
+ cv2.rectangle(annotated_img, (x, y), (x + w, y + h), box_color, kwargs.get("thickness", 1))
+
+ _draw_label(
+ annotated_img,
+ text=f"{roi_prob * 100:.1f}%",
+ x=x,
+ y=max(y - 6, 14),
+ prob_ratio=roi_prob,
+ )
+
+ return annotated_img
+
+ def display_bbox_from_array(self, face: np.ndarray, **kwargs) -> np.ndarray:
+ """얼굴 배열을 직접 입력받아 위조 의심 부위를 찾아 사각형 바운딩 박스와 확률을 그린다.
+
+ Args:
+ face (np.ndarray): 전처리(크롭)가 완료된 얼굴 이미지 배열 (RGB).
+ **kwargs:
+ category (int): 타겟 클래스 인덱스 (기본값: 1).
+ aug_smooth (bool): Test-time augmentation 가동 여부 (기본값: False).
+ eigen_smooth (bool): PCA 기반 노이즈 제거 가동 여부 (기본값: False).
+ threshold (float | str): 이진화 임계값 (기본값: "auto").
+ thickness (int): 사각형 선 굵기 (기본값: 2).
+
+ Returns:
+ np.ndarray: 바운딩 박스와 어노테이션이 추가된 원본 얼굴 이미지 (RGB).
+ """
+ cam_recovered, face = self._prepare_cam_from_array(
+ face,
+ category = kwargs.get("category", 1),
+ aug_smooth = kwargs.get("aug_smooth", False),
+ eigen_smooth = kwargs.get("eigen_smooth", False)
+ )
+
+ binary_mask = self._get_binary_mask(cam_recovered, kwargs.get("threshold", "auto"))
+
+ contours, _ = cv2.findContours(binary_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+
+ annotated_img = face.copy()
+ for c in contours:
+ x, y, w, h = cv2.boundingRect(c)
+ roi_prob = float(np.max(cam_recovered[y:y + h, x:x + w]))
+
+ r = int(min(255, 2 * roi_prob * 255))
+ g = int(min(255, 2 * (1 - roi_prob) * 255))
+ box_color = (r, g, 30) # RGB
+
+ cv2.rectangle(annotated_img, (x, y), (x + w, y + h), box_color, kwargs.get("thickness", 1))
+
+ _draw_label(
+ annotated_img,
+ text=f"{roi_prob * 100:.1f}%",
+ x=x,
+ y=max(y - 6, 14),
+ prob_ratio=roi_prob,
+ )
+
+ return annotated_img
\ No newline at end of file
diff --git a/explainability/eigenvalue_explainers.py b/explainability/explainer/eigenvalue_explainer.py
similarity index 69%
rename from explainability/eigenvalue_explainers.py
rename to explainability/explainer/eigenvalue_explainer.py
index 7f53174..9aa2633 100644
--- a/explainability/eigenvalue_explainers.py
+++ b/explainability/explainer/eigenvalue_explainer.py
@@ -1,5 +1,5 @@
-from pytorch_grad_cam import EigenCAM, EigenGradCAM, KPCA_CAM, SegEigenCAM
-from explainability.cam_explainer import CAMExplainer
+from pytorch_grad_cam import EigenCAM, EigenGradCAM, KPCA_CAM
+from explainability.explainer.cam_explainer import CAMExplainer
class EigenCAMExplainer(CAMExplainer):
"""
@@ -31,14 +31,5 @@ def _build_cam(self):
return KPCA_CAM(model=self.model, target_layers=self.target_layers, reshape_transform=self.reshape_fn,
kernel=self.kernel, gamma=self.gamma)
-class SegEigenCAMExplainer(CAMExplainer):
- """
- Like EigenCAM but with gradient weighting (absolute gradients ⊙ activations )
- before SVD and sign correction to fix SVD sign ambiguity
- designed for semantic segmentation
- """
- def _build_cam(self):
- return SegEigenCAM(model=self.model, target_layers=self.target_layers, reshape_transform=self.reshape_fn)
-
-
-
+ def __repr__(self):
+ return super().__repr__().rstrip(")") + f", kernel={self.kernel}, gamma={self.gamma})"
diff --git a/explainability/gradient_explainer.py b/explainability/explainer/gradient_explainer.py
similarity index 56%
rename from explainability/gradient_explainer.py
rename to explainability/explainer/gradient_explainer.py
index f50bff9..3767fc7 100644
--- a/explainability/gradient_explainer.py
+++ b/explainability/explainer/gradient_explainer.py
@@ -1,8 +1,8 @@
from pytorch_grad_cam import (
GradCAM, GradCAMPlusPlus, GradCAMElementWise,
- XGradCAM, HiResCAM, FullGrad, ShapleyCAM, FinerCAM,
+ XGradCAM, HiResCAM
)
-from explainability.cam_explainer import CAMExplainer
+from explainability.explainer.cam_explainer import CAMExplainer
class GradCAMExplainer(CAMExplainer):
@@ -44,29 +44,4 @@ def _build_cam(self):
return HiResCAM(model=self.model, target_layers=self.target_layers, reshape_transform=self.reshape_fn)
-class FullGradExplainer(CAMExplainer):
- """
- Computes the gradients of the biases from all over the network, and then sums them
- """
- def _build_cam(self):
- return FullGrad(model=self.model, target_layers=self.target_layers, reshape_transform=self.reshape_fn)
-
-
-class ShapleyCAMExplainer(CAMExplainer):
- """
- Weight the activations using the gradient and Hessian-vector product.
- """
- def _build_cam(self):
- return ShapleyCAM(model=self.model, target_layers=self.target_layers, reshape_transform=self.reshape_fn)
-
-
-class FinerCAMExplainer(CAMExplainer):
- """
- Improves fine-grained classification by comparing similar classes, suppressing shared features and highlighting discriminative details.
- """
- def __init__(self, base_method=GradCAM, **kwargs):
- self.base_method = base_method
- super().__init__(**kwargs)
-
- def _build_cam(self):
- return FinerCAM(model=self.model, target_layers=self.target_layers, reshape_transform=self.reshape_fn, base_method=self.base_method)
\ No newline at end of file
+
diff --git a/explainability/explainer/misc_explainer.py b/explainability/explainer/misc_explainer.py
new file mode 100644
index 0000000..85cb38f
--- /dev/null
+++ b/explainability/explainer/misc_explainer.py
@@ -0,0 +1,13 @@
+from pytorch_grad_cam import LayerCAM, RandomCAM
+from explainability.explainer.cam_explainer import CAMExplainer
+
+
+class LayerCAMExplainer(CAMExplainer):
+ """
+ Spatially weight the activations by positive gradients.
+ Works better especially in lower layers
+ """
+ def _build_cam(self):
+ return LayerCAM(model=self.model, target_layers=self.target_layers, reshape_transform=self.reshape_fn)
+
+
diff --git a/explainability/explainer/perturbation_explainer.py b/explainability/explainer/perturbation_explainer.py
new file mode 100644
index 0000000..cb460aa
--- /dev/null
+++ b/explainability/explainer/perturbation_explainer.py
@@ -0,0 +1,24 @@
+from pytorch_grad_cam import ScoreCAM, FEM
+from explainability.explainer.cam_explainer import CAMExplainer
+class ScoreCAMExplainer(CAMExplainer):
+ """
+ Perbutate the image by the scaled activation
+ mesaure how the output drops
+ """
+ def _build_cam(self):
+ return ScoreCAM(model=self.model, target_layers=self.target_layers, reshape_transform=self.reshape_fn)
+
+class FEMExplainer(CAMExplainer):
+ """
+ A gradient free method that binarizes activations
+ by an activation > mean + k * std rule
+ """
+ def __init__(self, k: int = 2, **kwargs):
+ self.k = k
+ super().__init__(**kwargs)
+
+ def _build_cam(self):
+ return FEM(model=self.model, target_layers=self.target_layers, reshape_transform=self.reshape_fn, k=self.k)
+
+ def __repr__(self):
+ return super().__repr__().rstrip(")") + f", k={self.k})"
\ No newline at end of file
diff --git a/explainability/metrics/__init__.py b/explainability/metrics/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/explainability/metrics/cam_mult_image.py b/explainability/metrics/cam_mult_image.py
new file mode 100644
index 0000000..c46751e
--- /dev/null
+++ b/explainability/metrics/cam_mult_image.py
@@ -0,0 +1,37 @@
+import torch
+import numpy as np
+from typing import List, Callable
+from explainability.metrics.perturbation_confidence import PerturbationConfidenceMetric
+
+
+def multiply_tensor_with_cam(input_tensor: torch.Tensor,
+ cam: torch.Tensor):
+ """ Multiply an input tensor (after normalization)
+ with a pixel attribution map
+ """
+ return input_tensor * cam
+
+
+class CamMultImageConfidenceChange(PerturbationConfidenceMetric):
+ def __init__(self):
+ super(CamMultImageConfidenceChange,
+ self).__init__(multiply_tensor_with_cam)
+
+
+class DropInConfidence(CamMultImageConfidenceChange):
+ def __init__(self):
+ super(DropInConfidence, self).__init__()
+
+ def __call__(self, *args, **kwargs):
+ scores = super(DropInConfidence, self).__call__(*args, **kwargs)
+ scores = -scores
+ return np.maximum(scores, 0)
+
+
+class IncreaseInConfidence(CamMultImageConfidenceChange):
+ def __init__(self):
+ super(IncreaseInConfidence, self).__init__()
+
+ def __call__(self, *args, **kwargs):
+ scores = super(IncreaseInConfidence, self).__call__(*args, **kwargs)
+ return np.float32(scores > 0)
\ No newline at end of file
diff --git a/explainability/metrics/perturbation_confidence.py b/explainability/metrics/perturbation_confidence.py
new file mode 100644
index 0000000..211ab8a
--- /dev/null
+++ b/explainability/metrics/perturbation_confidence.py
@@ -0,0 +1,109 @@
+import torch
+import numpy as np
+from typing import List, Callable
+
+import numpy as np
+import cv2
+
+
+class PerturbationConfidenceMetric:
+ def __init__(self, perturbation):
+ self.perturbation = perturbation
+
+ def __call__(self, input_tensor: torch.Tensor,
+ cams: np.ndarray,
+ targets: List[Callable],
+ model: torch.nn.Module,
+ return_visualization=False,
+ return_diff=True):
+
+ if return_diff:
+ with torch.no_grad():
+ outputs = model(input_tensor)
+ scores = [target(output).cpu().numpy()
+ for target, output in zip(targets, outputs)]
+ scores = np.float32(scores)
+
+ batch_size = input_tensor.size(0)
+ perturbated_tensors = []
+ for i in range(batch_size):
+ cam = cams[i]
+ tensor = self.perturbation(input_tensor[i, ...].cpu(),
+ torch.from_numpy(cam))
+ tensor = tensor.to(input_tensor.device)
+ perturbated_tensors.append(tensor.unsqueeze(0))
+ perturbated_tensors = torch.cat(perturbated_tensors)
+
+ with torch.no_grad():
+ outputs_after_imputation = model(perturbated_tensors)
+ scores_after_imputation = [
+ target(output).cpu().numpy() for target, output in zip(
+ targets, outputs_after_imputation)]
+ scores_after_imputation = np.float32(scores_after_imputation)
+
+ if return_diff:
+ result = scores_after_imputation - scores
+ else:
+ result = scores_after_imputation
+
+ if return_visualization:
+ return result, perturbated_tensors
+ else:
+ return result
+
+
+class RemoveMostRelevantFirst:
+ def __init__(self, percentile, imputer):
+ self.percentile = percentile
+ self.imputer = imputer
+
+ def __call__(self, input_tensor, mask):
+ imputer = self.imputer
+ if self.percentile != 'auto':
+ threshold = np.percentile(mask.cpu().numpy(), self.percentile)
+ binary_mask = np.float32(mask < threshold)
+ else:
+ _, binary_mask = cv2.threshold(
+ np.uint8(mask * 255), 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
+
+ binary_mask = torch.from_numpy(binary_mask)
+ binary_mask = binary_mask.to(mask.device)
+ return imputer(input_tensor, binary_mask)
+
+
+class RemoveLeastRelevantFirst(RemoveMostRelevantFirst):
+ def __init__(self, percentile, imputer):
+ super(RemoveLeastRelevantFirst, self).__init__(percentile, imputer)
+
+ def __call__(self, input_tensor, mask):
+ return super(RemoveLeastRelevantFirst, self).__call__(
+ input_tensor, 1 - mask)
+
+
+class AveragerAcrossThresholds:
+ def __init__(
+ self,
+ imputer,
+ percentiles=[
+ 10,
+ 20,
+ 30,
+ 40,
+ 50,
+ 60,
+ 70,
+ 80,
+ 90]):
+ self.imputer = imputer
+ self.percentiles = percentiles
+
+ def __call__(self,
+ input_tensor: torch.Tensor,
+ cams: np.ndarray,
+ targets: List[Callable],
+ model: torch.nn.Module):
+ scores = []
+ for percentile in self.percentiles:
+ imputer = self.imputer(percentile)
+ scores.append(imputer(input_tensor, cams, targets, model))
+ return np.mean(np.float32(scores), axis=0)
\ No newline at end of file
diff --git a/explainability/metrics/road.py b/explainability/metrics/road.py
new file mode 100644
index 0000000..905f84e
--- /dev/null
+++ b/explainability/metrics/road.py
@@ -0,0 +1,154 @@
+import torch
+import numpy as np
+from scipy.sparse import lil_matrix, csc_matrix
+from scipy.sparse.linalg import spsolve
+from typing import List, Callable
+from explainability.metrics.perturbation_confidence import (
+ PerturbationConfidenceMetric, AveragerAcrossThresholds,
+ RemoveMostRelevantFirst, RemoveLeastRelevantFirst
+)
+
+# The weights of the surrounding pixels
+neighbors_weights = [((1, 1), 1 / 12),
+ ((0, 1), 1 / 6),
+ ((-1, 1), 1 / 12),
+ ((1, -1), 1 / 12),
+ ((0, -1), 1 / 6),
+ ((-1, -1), 1 / 12),
+ ((1, 0), 1 / 6),
+ ((-1, 0), 1 / 6)]
+
+
+class NoisyLinearImputer:
+ def __init__(self,
+ noise: float = 0.01,
+ weighting: List[float] = neighbors_weights):
+ """
+ Noisy linear imputation.
+ noise: magnitude of noise to add (absolute, set to 0 for no noise)
+ weighting: Weights of the neighboring pixels in the computation.
+ List of tuples of (offset, weight)
+ """
+ self.noise = noise
+ self.weighting = neighbors_weights
+
+ @staticmethod
+ def add_offset_to_indices(indices, offset, mask_shape):
+ """ Add the corresponding offset to the indices.
+ Return new indices plus a valid bit-vector. """
+ cord1 = indices % mask_shape[1]
+ cord0 = indices // mask_shape[1]
+ cord0 += offset[0]
+ cord1 += offset[1]
+ valid = ((cord0 < 0) | (cord1 < 0) |
+ (cord0 >= mask_shape[0]) |
+ (cord1 >= mask_shape[1]))
+ return ~valid, indices + offset[0] * mask_shape[1] + offset[1]
+
+ @staticmethod
+ def setup_sparse_system(mask, img, neighbors_weights):
+ """ Vectorized version to set up the equation system.
+ mask: (H, W)-tensor of missing pixels.
+ Image: (H, W, C)-tensor of all values.
+ Return (N,N)-System matrix, (N,C)-Right hand side for each of the C channels.
+ """
+ maskflt = mask.flatten()
+ imgflat = img.reshape((img.shape[0], -1))
+ # Indices that are imputed in the flattened mask:
+ indices = np.argwhere(maskflt == 0).flatten()
+ coords_to_vidx = np.zeros(len(maskflt), dtype=int)
+ coords_to_vidx[indices] = np.arange(len(indices))
+ numEquations = len(indices)
+ # System matrix:
+ A = lil_matrix((numEquations, numEquations))
+ b = np.zeros((numEquations, img.shape[0]))
+ # Sum of weights assigned:
+ sum_neighbors = np.ones(numEquations)
+ for n in neighbors_weights:
+ offset, weight = n[0], n[1]
+ # Take out outliers
+ valid, new_coords = NoisyLinearImputer.add_offset_to_indices(
+ indices, offset, mask.shape)
+ valid_coords = new_coords[valid]
+ valid_ids = np.argwhere(valid == 1).flatten()
+ # Add values to the right hand-side
+ has_values_coords = valid_coords[maskflt[valid_coords] > 0.5]
+ has_values_ids = valid_ids[maskflt[valid_coords] > 0.5]
+ b[has_values_ids, :] -= weight * imgflat[:, has_values_coords].T
+ # Add weights to the system (left hand side)
+# Find coordinates in the system.
+ has_no_values = valid_coords[maskflt[valid_coords] < 0.5]
+ variable_ids = coords_to_vidx[has_no_values]
+ has_no_values_ids = valid_ids[maskflt[valid_coords] < 0.5]
+ A[has_no_values_ids, variable_ids] = weight
+ # Reduce weight for invalid
+ sum_neighbors[np.argwhere(valid == 0).flatten()] = \
+ sum_neighbors[np.argwhere(valid == 0).flatten()] - weight
+
+ A[np.arange(numEquations), np.arange(numEquations)] = -sum_neighbors
+ return A, b
+
+ def __call__(self, img: torch.Tensor, mask: torch.Tensor):
+ """ Our linear inputation scheme. """
+ """
+ This is the function to do the linear infilling
+ img: original image (C,H,W)-tensor;
+ mask: mask; (H,W)-tensor
+
+ """
+ imgflt = img.reshape(img.shape[0], -1)
+ maskflt = mask.reshape(-1)
+ # Indices that need to be imputed.
+ indices_linear = np.argwhere(maskflt == 0).flatten()
+ # Set up sparse equation system, solve system.
+ A, b = NoisyLinearImputer.setup_sparse_system(
+ mask.numpy(), img.numpy(), neighbors_weights)
+ res = torch.tensor(spsolve(csc_matrix(A), b), dtype=torch.float)
+
+ # Fill the values with the solution of the system.
+ img_infill = imgflt.clone()
+ img_infill[:, indices_linear] = res.t() + self.noise * \
+ torch.randn_like(res.t())
+
+ return img_infill.reshape_as(img)
+
+
+class ROADMostRelevantFirst(PerturbationConfidenceMetric):
+ def __init__(self, percentile=80):
+ super(ROADMostRelevantFirst, self).__init__(
+ RemoveMostRelevantFirst(percentile, NoisyLinearImputer()))
+
+
+class ROADLeastRelevantFirst(PerturbationConfidenceMetric):
+ def __init__(self, percentile=20):
+ super(ROADLeastRelevantFirst, self).__init__(
+ RemoveLeastRelevantFirst(percentile, NoisyLinearImputer()))
+
+
+class ROADMostRelevantFirstAverage(AveragerAcrossThresholds):
+ def __init__(self, percentiles=[10, 20, 30, 40, 50, 60, 70, 80, 90]):
+ super(ROADMostRelevantFirstAverage, self).__init__(
+ ROADMostRelevantFirst, percentiles)
+
+
+class ROADLeastRelevantFirstAverage(AveragerAcrossThresholds):
+ def __init__(self, percentiles=[10, 20, 30, 40, 50, 60, 70, 80, 90]):
+ super(ROADLeastRelevantFirstAverage, self).__init__(
+ ROADLeastRelevantFirst, percentiles)
+
+
+class ROADCombined:
+ def __init__(self, percentiles=[10, 20, 30, 40, 50, 60, 70, 80, 90]):
+ self.percentiles = percentiles
+ self.morf_averager = ROADMostRelevantFirstAverage(percentiles)
+ self.lerf_averager = ROADLeastRelevantFirstAverage(percentiles)
+
+ def __call__(self,
+ input_tensor: torch.Tensor,
+ cams: np.ndarray,
+ targets: List[Callable],
+ model: torch.nn.Module):
+
+ scores_lerf = self.lerf_averager(input_tensor, cams, targets, model)
+ scores_morf = self.morf_averager(input_tensor, cams, targets, model)
+ return (scores_lerf - scores_morf) / 2
\ No newline at end of file
diff --git a/explainability/utils/image.py b/explainability/utils/image.py
index 83b90fa..92a5ba4 100644
--- a/explainability/utils/image.py
+++ b/explainability/utils/image.py
@@ -44,4 +44,18 @@ def deprocess_image(tensor):
img = (img * std) + mean
img = np.clip(img, 0, 1) # range(0~1)
- return np.float32(img)
\ No newline at end of file
+ return np.float32(img)
+
+def remove_padding_and_resize(cam, orig_shape, target_shape):
+ orig_h, orig_w = orig_shape[:2] # (H,W,3)
+ target_h, target_w = target_shape[2:] # (1,3,H,W)
+
+ scale = max(target_h, target_w) / max(orig_h, orig_w)
+ new_h, new_w = int(orig_h * scale), int(orig_w * scale)
+
+ pad_h = (target_h - new_h) // 2
+ pad_w = (target_w - new_w) // 2
+
+ # Remove Padding
+ unpadded_cam = cam[pad_h:pad_h+new_h, pad_w:pad_w+new_w]
+ return cv2.resize(unpadded_cam, (orig_w, orig_h))
\ No newline at end of file
diff --git a/explainability/utils/model_targets.py b/explainability/utils/model_targets.py
index 62ba03a..ba63a85 100644
--- a/explainability/utils/model_targets.py
+++ b/explainability/utils/model_targets.py
@@ -1,3 +1,5 @@
+import torch
+
class BinaryClassifierOutputTarget:
def __init__(self, category):
self.category = category
@@ -8,4 +10,7 @@ def __call__(self, model_output):
else:
sign = -1
return model_output * sign
-
\ No newline at end of file
+
+class BinaryClassifierOutputSigmoidTarget:
+ def __call__(self, model_output):
+ return torch.sigmoid(model_output)
\ No newline at end of file
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..2d4ad22
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,51 @@
+[build-system]
+requires = ["setuptools>=68", "wheel"]
+build-backend = "setuptools.build_meta"
+
+# ─────────────────────────────────────────────
+# [project]
+# PyPI에 올라가는 패키지 메타데이터
+# pip install deepguard 했을 때 보이는 정보들
+# ─────────────────────────────────────────────
+[project]
+name = "deepguard" # pip install deepguard 할 때 이 이름
+version = "0.2.1" # pip install deepguard==0.2.1 으로 버전 고정 가능
+description = "Multi-Scale Efficient Vision Transformer for Robust Deepfake Detection"
+readme = "README.md" # PyPI 페이지에 표시될 설명 (마크다운)
+requires-python = ">=3.10" # 이 버전 미만이면 pip install 자체를 막음
+license = { text = "MIT" }
+authors = [
+ { name = "seoyunje", email = "seoyunje2001@gmail.com" }
+]
+classifiers = [
+ "Programming Language :: Python :: 3",
+ "Programming Language :: Python :: 3.10",
+ "Programming Language :: Python :: 3.11",
+ "License :: OSI Approved :: MIT License",
+ "Operating System :: OS Independent",
+ "Topic :: Scientific/Engineering :: Artificial Intelligence",
+]
+
+dependencies = ["torch>=2.10.0",
+ "torchmetrics>=1.6.1",
+ "timm>=1.0.12",
+ "ultralytics>=8.4.41",
+ "grad-cam>=1.5.5"
+ ]
+
+[project.urls]
+Homepage = "https://github.com/HanMoonSub/DeepGuard"
+"Bug Tracker" = "https://github.com/HanMoonSub/DeepGuard/issues"
+"Source Code" = "https://github.com/HanMoonSub/DeepGuard"
+
+
+[tool.setuptools.packages.find]
+
+include = [
+ "deepguard*", # 모델, 레이어, 데이터셋 등 핵심 패키지
+ "inference*", # ImagePredictor, VideoPredictor
+ "explainability*", # CAM 기반 XAI
+ "preprocess*", # 얼굴 검출, 프레임 추출 등
+]
+
+exclude = ["App*", "client*"]
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
index ce368d3..a7bc9ab 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -5,6 +5,7 @@ python-multipart>=0.0.22
pydantic[email]>=2.12.5
aiofiles>=25.1.0
itsdangerous>=2.2.0
+python-dotenv>=0.9.9
# --- [Database & ORM] ---
sqlalchemy>=2.0.48
@@ -16,6 +17,7 @@ redis>=7.4.0
# --- [Security & Auth] ---
bcrypt>=4.0.1
passlib[bcrypt]>=1.7.4
+cryptography>=42.0.0
# --- [Data Analysis & Math] ---
numpy>=1.23.0,<3.0.0
@@ -40,4 +42,7 @@ seaborn>=0.13.2
colorama>=0.4.6
# --- [Distributed Task Queue] ---
-celery>=5.6.3
\ No newline at end of file
+celery>=5.6.3
+
+# --- [Linting & Formatting] ---
+ruff>=0.9.0
\ No newline at end of file
diff --git a/setup.py b/setup.py
deleted file mode 100644
index 16d3c3c..0000000
--- a/setup.py
+++ /dev/null
@@ -1,30 +0,0 @@
-import setuptools
-
-with open("README.md", mode='r', encoding='utf-8') as fh:
- long_description = fh.read()
-
-setuptools.setup(
- name = 'deepguard',
- version = '0.2.1',
- author = 'seoyunje',
- author_email = "seoyunje2001@gmail.com",
- description = "Multi-Scale Efficient Vision Transformer for Robust Deepfake Detection",
- long_description = long_description,
- long_description_content_type ='text/markdown',
- url = 'https://github.com/HanMoonSub/DeepGuard',
- project_urls = {
- "Bug Tracker": "https://github.com/HanMoonSub/DeepGuard/issues",
- "Source Code": "https://github.com/HanMoonSub/DeepGuard",
- },
- classifiers = [
- 'Programming Language :: Python :: 3',
- 'Programming Language :: Python :: 3.10',
- 'Programming Language :: Python :: 3.11',
- 'License :: OSI Approved :: MIT License',
- 'Operating System :: OS Independent',
- 'Topic :: Scientific/Engineering :: Artificial Intelligence',
- ],
- packages = setuptools.find_packages(exclude=['Attention', 'preprocess','inference']),
- python_requires = '>=3.10',
- install_requires = []
-)
\ No newline at end of file