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 -

- DeepGuard Banner -

-

License Stars + Downloads + Last Commit Status Release + Repo Size +

+ +

+ DeepGuard Banner

@@ -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 | +| :--- | :--- | :--- | +| ![](https://img.shields.io/badge/Low_level-blue?style=flat-square) | **HiResCAM** | Like GradCAM but element-wise multiply the activations with the gradients; provably guaranteed faithfulness for certain models | +| ![](https://img.shields.io/badge/Low_level-blue?style=flat-square) | **GradCAMElementWise** | Like GradCAM but element-wise multiply the activations with the gradients then apply a ReLU operation before summing | +| ![](https://img.shields.io/badge/Low_level-blue?style=flat-square) | **LayerCAM** | Spatially weight the activations by positive gradients. Works better especially in lower layers | +| --- | --- | --- | --- | +| ![](https://img.shields.io/badge/High_level-red?style=flat-square) | **EigenGradCAM** | Like EigenCAM but with class discrimination: First principle component of Activations*Grad. Looks like GradCAM, but cleaner | +| ![](https://img.shields.io/badge/High_level-red?style=flat-square) | **GradCAM++** | Like GradCAM but uses second order gradients | +| ![](https://img.shields.io/badge/High_level-red?style=flat-square) | **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** | ![](https://img.shields.io/badge/Low_level_Branch-blue?style=flat-square) | | | | | +| **⚡ ms-eff-vit-b5** | ![](https://img.shields.io/badge/Low_level_Branch-blue?style=flat-square) | | | | | + +| Model | Branch-Level | Image | EigenGradCam | GradCamPlusPlus | XGradCam | +| :--- | :---: | :---: | :---: | :---: | :---: | +| **⚡ ms-eff-vit-b0** | ![](https://img.shields.io/badge/High_level_Branch-red?style=flat-square) | | | | | +| **🔥 ms-eff-vit-b5** | ![](https://img.shields.io/badge/High_level_Branch-red?style=flat-square) | | | | | + + + +## 📊 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