Deploy A-EYE remote full ensemble backend

.dockerignore

@@ -3,9 +3,14 @@ __pycache__/
 *.pyc
 .pytest_cache/
 .env
-models/*.pt
-models/*.pth
-models/*.onnx
-models/*.safetensors
 *.log
 .DS_Store
+experiments/
+dataset/
+dataset_20260509/
+hard_examples/
+logs/
+_hf_space_ensemble/
+_hf_space_remote_full/
+_hf_space_slim_32_37/
+_hf_model_repo_full_ensemble/

.env.example

@@ -1,4 +0,0 @@
-MODEL_PATH=./models/best.pt
-INPUT_SIZE=224
-IMAGE_MEAN=0.485,0.456,0.406
-IMAGE_STD=0.229,0.224,0.225

.gitignore

@@ -1,11 +0,0 @@
-venv/
-__pycache__/
-*.pyc
-.pytest_cache/
-.env
-models/*.pt
-models/*.pth
-models/*.onnx
-models/*.safetensors
-*.log
-.DS_Store

DEPLOY.md

@@ -1,95 +0,0 @@
-# A-EYE 백엔드 무료 배포 가이드 (Hugging Face Spaces)
-
-앱이 **앱스토어 심사를 통과하려면**, 심사관이 어디서든 접속할 수 있는
-**공개 HTTPS 백엔드**가 있어야 합니다. (집 와이파이 IP나 ngrok 임시 주소로는 통과 못 합니다.)
-
-이 모델은 **737MB + torch** 라서 RAM 2GB 이상이 필요한데,
-**Hugging Face Spaces** 가 RAM 16GB를 무료로 제공해서 유일하게 맞는 무료 옵션입니다.
-(Render 무료 512MB는 메모리 부족으로 실패합니다.)
-
----
-
-## 1단계 — 모델 파일 업로드 (브라우저, 터미널 X)
-
-1. https://huggingface.co 가입 (구글 계정으로 30초).
-2. 우측 상단 프로필 → **New Model** → 이름 예: `aeye-model` → Create.
-3. 모델 페이지 → **Files and versions** 탭 → **Add file → Upload files** →
-   `best.pt` 드래그&드롭 → Commit. (대용량 LFS 자동 처리)
-4. 업로드된 `best.pt` 옆 **⋯ → Copy download link** 로 직링크 복사. 형식:
-   ```
-   https://huggingface.co/<유저명>/aeye-model/resolve/main/best.pt
-   ```
-   이게 아래에서 쓸 **MODEL_URL** 입니다.
-
----
-
-## 2단계 — Space(서버) 만들기 (브라우저)
-
-1. 프로필 → **New Space**.
-2. 설정:
-   - **Owner / Space name**: 예 `aeye-backend`
-   - **SDK**: **Docker** 선택 (중요)
-   - **Hardware**: `CPU basic` (무료, 16GB RAM)
-   - 공개/비공개: **Public** 권장 (앱이 호출해야 함)
-3. **Create Space** → 빈 Space가 생깁니다.
-
----
-
-## 3단계 — 백엔드 코드 올리기 (터미널, 명령 몇 줄)
-
-Space는 git 저장소입니다. `a-eye/backend` 폴더 내용을 Space에 넣습니다.
-
-```bash
-# 작업 폴더에서
-git clone https://huggingface.co/spaces/<유저명>/aeye-backend
-cp -R "/Users/jun/Desktop/a-eye Design System/a-eye/backend/." aeye-backend/
-cd aeye-backend
-git add .
-git commit -m "deploy A-EYE backend"
-git push
-```
-
-> push 할 때 사용자명은 HF 아이디, 비밀번호는 **HF 액세스 토큰**입니다.
-> 토큰: https://huggingface.co/settings/tokens → New token(write) 생성 후 비밀번호 자리에 붙여넣기.
-> (`venv/`, 모델 파일은 `.gitignore`로 자동 제외됩니다.)
-
----
-
-## 4단계 — 모델 주소 알려주기 (브라우저)
-
-1. Space 페이지 → **Settings → Variables and secrets** → **New secret**.
-2. 추가:
-   | Name | Value |
-   |------|-------|
-   | `MODEL_URL` | 1단계에서 복사한 `best.pt` 직링크 |
-
-3. Space가 자동으로 다시 빌드됩니다. 첫 빌드는 torch 설치 + 모델 다운로드로 **5~10분** 걸립니다.
-
-### 동작 확인
-Space 주소는 `https://<유저명>-aeye-backend.hf.space` 형태입니다.
-```bash
-curl https://<유저명>-aeye-backend.hf.space/api/health     # {"status":"ok"}
-curl https://<유저명>-aeye-backend.hf.space/api/version    # model_version 표시(mock 아니면 정상)
-```
-
----
-
-## 5단계 — 앱에 백엔드 주소 연결
-
-`a-eye/app.json` 의 `extra.defaultBackendUrl` 수정:
-```json
-"extra": {
-  "defaultBackendUrl": "https://<유저명>-aeye-backend.hf.space"
-}
-```
-이제 앱은 설치 즉시 분석이 동작합니다. (사용자는 설정에서 다른 서버로 바꿀 수도 있음)
-
----
-
-## 참고
-- 무료 Space는 일정 시간 미사용 시 잠들고(sleep), 다음 첫 요청 때 깨어나며 30초~1분 걸릴 수 있습니다.
-  앱스토어 심사 직전에 위 `curl /api/health` 로 한 번 깨워두면 좋습니다.
-- 트래픽이 늘면 Space Settings에서 유료 하드웨어로 업그레이드할 수 있습니다.
-
-> Render/Railway/Fly.io 도 같은 `Dockerfile` + `MODEL_URL` 환경변수로 배포할 수 있습니다.
-> 단, 모델 로딩에 RAM 2GB 이상 플랜이 필요하며 대부분 유료입니다.

Dockerfile

@@ -16,6 +16,7 @@ RUN apt-get update \
 
 COPY requirements.txt .
 RUN pip install --upgrade pip \
+    && pip install torch==2.5.1 torchvision==0.20.1 --index-url https://download.pytorch.org/whl/cpu \
     && pip install -r requirements.txt
 
 COPY . .

README.md

@@ -1,54 +1,23 @@
 ---
 title: A-EYE Backend
-emoji: 🧠
-colorFrom: indigo
-colorTo: blue
 sdk: docker
 app_port: 8000
 pinned: false
 ---
 
-# A-EYE 백엔드
+# A-EYE Backend
 
-FastAPI 기반 이미지 판별 서버입니다. `MODEL_PATH` 파일이 없거나 비어 있으면 MOCK 모드로 실행되어 앱 end-to-end 테스트가 가능합니다.
+Production ensemble API for A-EYE.
 
-## 실행
+Health:
 
-```bash
-python -m venv venv && source venv/bin/activate
-pip install -r requirements.txt
-cp .env.example .env
-uvicorn main:app --host 0.0.0.0 --port 8000 --reload
+```text
+GET /api/health
 ```
 
-## 팀원 모델 파일 적용
+Prediction:
 
-구글드라이브에서 받은 `best22.pt`는 GitHub에 올리지 않습니다. 각자 내려받은 뒤 파일 이름을 `best.pt`로 바꿔서 아래 위치에 넣어 주세요.
-
-```bash
-mkdir -p models
-cp ~/Downloads/best22.pt models/best.pt
-echo "24_realworld_artifact_30ep" > models/best.pt.version.txt
-```
-
-`.env`의 `MODEL_PATH`는 기본값 그대로 두면 됩니다.
-
-```bash
-MODEL_PATH=./models/best.pt
+```text
+POST /api/predict
+multipart form field: image
 ```
-
-## 모델 교체
-
-1. `models/best.pt`를 실제 모델 파일로 교체합니다.
-2. 선택 사항으로 `models/best.pt.version.txt`의 버전을 갱신합니다.
-3. `uvicorn`을 재시작합니다.
-
-코드 수정은 필요 없습니다.
-
-## GPU 사용
-
-기본은 CPU 로딩입니다. GPU 서버에서는 `model_loader.py`의 `map_location`을 환경에 맞춰 조정하거나 배포 이미지에서 CUDA PyTorch를 설치하세요.
-
-## CORS
-
-개발 편의를 위해 모든 origin을 허용합니다. 운영 환경에서는 `main.py`의 `allow_origins`를 앱 도메인 또는 배포된 클라이언트 주소로 좁히세요.

artifact_model.py

@@ -1,273 +0,0 @@
-from collections import OrderedDict
-from typing import Any
-
-import torch
-from torch import nn
-from torchvision.models import vit_b_16
-
-
-class LoRALinear(nn.Module):
-    def __init__(self, in_features: int, out_features: int, r: int = 32, alpha: int = 64, bias: bool = True) -> None:
-        super().__init__()
-        self.base = nn.Linear(in_features, out_features, bias=bias)
-        self.lora_A = nn.Parameter(torch.zeros(r, in_features))
-        self.lora_B = nn.Parameter(torch.zeros(out_features, r))
-        self.scale = alpha / r
-
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        return self.base(x) + torch.nn.functional.linear(
-            torch.nn.functional.linear(x, self.lora_A),
-            self.lora_B,
-        ) * self.scale
-
-
-class ViTEncoder(nn.Module):
-    def __init__(self, dropout: float = 0.32) -> None:
-        super().__init__()
-        self.backbone = vit_b_16(weights=None)
-        self.backbone.heads = nn.Identity()
-        for layer in self.backbone.encoder.layers:
-            layer.mlp[0] = LoRALinear(768, 3072)
-            layer.mlp[3] = LoRALinear(3072, 768)
-        self.head = nn.Sequential(
-            nn.BatchNorm1d(768),
-            nn.Dropout(dropout),
-            nn.Linear(768, 2),
-        )
-
-    def features(self, x: torch.Tensor) -> torch.Tensor:
-        return self.backbone(x)
-
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        return self.head(self.features(x))
-
-
-class CLIPSelfAttention(nn.Module):
-    def __init__(self) -> None:
-        super().__init__()
-        self.k_proj = nn.Linear(768, 768)
-        self.v_proj = LoRALinear(768, 768)
-        self.q_proj = LoRALinear(768, 768)
-        self.out_proj = nn.Linear(768, 768)
-        self.num_heads = 12
-        self.head_dim = 64
-        self.scale = self.head_dim ** -0.5
-
-    def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:
-        batch, tokens, _ = hidden_states.shape
-        q = self.q_proj(hidden_states).view(batch, tokens, self.num_heads, self.head_dim).transpose(1, 2)
-        k = self.k_proj(hidden_states).view(batch, tokens, self.num_heads, self.head_dim).transpose(1, 2)
-        v = self.v_proj(hidden_states).view(batch, tokens, self.num_heads, self.head_dim).transpose(1, 2)
-        attn = torch.softmax((q @ k.transpose(-2, -1)) * self.scale, dim=-1)
-        out = (attn @ v).transpose(1, 2).reshape(batch, tokens, 768)
-        return self.out_proj(out)
-
-
-class CLIPEncoderLayer(nn.Module):
-    def __init__(self) -> None:
-        super().__init__()
-        self.self_attn = CLIPSelfAttention()
-        self.layer_norm1 = nn.LayerNorm(768)
-        self.mlp = nn.Module()
-        self.mlp.fc1 = nn.Linear(768, 3072)
-        self.mlp.fc2 = nn.Linear(3072, 768)
-        self.layer_norm2 = nn.LayerNorm(768)
-        self.activation = nn.GELU()
-
-    def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:
-        hidden_states = hidden_states + self.self_attn(self.layer_norm1(hidden_states))
-        mlp_out = self.mlp.fc2(self.activation(self.mlp.fc1(self.layer_norm2(hidden_states))))
-        return hidden_states + mlp_out
-
-
-class CLIPVisionInner(nn.Module):
-    def __init__(self) -> None:
-        super().__init__()
-        self.embeddings = nn.Module()
-        self.embeddings.class_embedding = nn.Parameter(torch.zeros(768))
-        self.embeddings.patch_embedding = nn.Conv2d(3, 768, kernel_size=32, stride=32, bias=False)
-        self.embeddings.position_embedding = nn.Embedding(50, 768)
-        self.pre_layrnorm = nn.LayerNorm(768)
-        self.encoder = nn.Module()
-        self.encoder.layers = nn.ModuleList([CLIPEncoderLayer() for _ in range(12)])
-        self.post_layernorm = nn.LayerNorm(768)
-
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        patch = self.embeddings.patch_embedding(x).flatten(2).transpose(1, 2)
-        cls = self.embeddings.class_embedding.to(dtype=patch.dtype).expand(x.shape[0], 1, -1)
-        hidden = torch.cat([cls, patch], dim=1)
-        positions = torch.arange(hidden.shape[1], device=x.device).unsqueeze(0)
-        hidden = hidden + self.embeddings.position_embedding(positions)
-        hidden = self.pre_layrnorm(hidden)
-        for layer in self.encoder.layers:
-            hidden = layer(hidden)
-        return self.post_layernorm(hidden[:, 0])
-
-
-class CLIPVisionWithProjection(nn.Module):
-    def __init__(self) -> None:
-        super().__init__()
-        self.vision_model = CLIPVisionInner()
-        self.visual_projection = nn.Linear(768, 512, bias=False)
-
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        return self.visual_projection(self.vision_model(x))
-
-
-class CLIPEncoder(nn.Module):
-    def __init__(self, dropout: float = 0.32) -> None:
-        super().__init__()
-        self.vision_model = CLIPVisionWithProjection()
-        self.head = nn.Sequential(
-            nn.BatchNorm1d(512),
-            nn.Dropout(dropout),
-            nn.Linear(512, 2),
-        )
-
-    def features(self, x: torch.Tensor) -> torch.Tensor:
-        return self.vision_model(x)
-
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        return self.head(self.features(x))
-
-
-class FFTBranch(nn.Module):
-    def __init__(self, dropout: float = 0.10) -> None:
-        super().__init__()
-        self.conv = nn.Sequential(OrderedDict([
-            ("0", nn.Conv2d(3, 32, 7, padding=3, bias=False)),
-            ("1", nn.BatchNorm2d(32)),
-            ("2", nn.ReLU(inplace=True)),
-            ("3", nn.Conv2d(32, 64, 5, padding=2, bias=False)),
-            ("4", nn.BatchNorm2d(64)),
-            ("5", nn.ReLU(inplace=True)),
-            ("6", nn.Conv2d(64, 128, 3, padding=1, bias=False)),
-            ("7", nn.BatchNorm2d(128)),
-            ("8", nn.ReLU(inplace=True)),
-            ("9", nn.Conv2d(128, 128, 3, padding=1, bias=False)),
-            ("10", nn.BatchNorm2d(128)),
-            ("11", nn.ReLU(inplace=True)),
-        ]))
-        self.pool = nn.AdaptiveAvgPool2d((1, 1))
-        self.proj = nn.Sequential(
-            nn.Dropout(dropout),
-            nn.Linear(128, 256),
-            nn.LayerNorm(256),
-        )
-
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        fft = torch.fft.fft2(x, norm="ortho")
-        mag = torch.log1p(torch.abs(torch.fft.fftshift(fft, dim=(-2, -1))))
-        feat = self.pool(self.conv(mag)).flatten(1)
-        return self.proj(feat)
-
-
-class SRMBranch(nn.Module):
-    def __init__(self, dropout: float = 0.10) -> None:
-        super().__init__()
-        self.register_buffer("clip_mean", torch.zeros(1, 3, 1, 1))
-        self.register_buffer("clip_std", torch.ones(1, 3, 1, 1))
-        self.highpass = nn.Conv2d(3, 18, 5, padding=2, bias=False)
-        self.encoder = nn.Sequential(OrderedDict([
-            ("0", nn.Conv2d(18, 48, 3, padding=1, bias=False)),
-            ("1", nn.BatchNorm2d(48)),
-            ("2", nn.ReLU(inplace=True)),
-            ("3", nn.Conv2d(48, 96, 3, padding=1, bias=False)),
-            ("4", nn.BatchNorm2d(96)),
-            ("5", nn.ReLU(inplace=True)),
-            ("6", nn.Conv2d(96, 128, 3, padding=1, bias=False)),
-            ("7", nn.BatchNorm2d(128)),
-            ("8", nn.ReLU(inplace=True)),
-            ("9", nn.Conv2d(128, 128, 3, padding=1, bias=False)),
-            ("10", nn.BatchNorm2d(128)),
-            ("11", nn.ReLU(inplace=True)),
-        ]))
-        self.pool = nn.AdaptiveAvgPool2d((1, 1))
-        self.proj = nn.Sequential(
-            nn.Dropout(dropout),
-            nn.Linear(128, 128),
-            nn.LayerNorm(128),
-        )
-
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        normalized = (x - self.clip_mean) / self.clip_std.clamp_min(1e-6)
-        residual = self.highpass(normalized)
-        feat = self.pool(self.encoder(residual)).flatten(1)
-        return self.proj(feat)
-
-
-class ArtifactLoRAHybrid(nn.Module):
-    expects_raw_input = True
-
-    def __init__(self, dropout: float = 0.32) -> None:
-        super().__init__()
-        self.register_buffer("clip_mean", torch.zeros(1, 3, 1, 1))
-        self.register_buffer("clip_std", torch.ones(1, 3, 1, 1))
-        self.vit_encoder = ViTEncoder(dropout=dropout)
-        self.clip_encoder = CLIPEncoder(dropout=dropout)
-        self.fft_branch = FFTBranch()
-        self.srm_branch = SRMBranch()
-        self.head = nn.Sequential(
-            nn.BatchNorm1d(1664),
-            nn.Linear(1664, 832),
-            nn.ReLU(inplace=True),
-            nn.Dropout(dropout),
-            nn.Linear(832, 416),
-            nn.ReLU(inplace=True),
-            nn.Dropout(dropout),
-            nn.Linear(416, 2),
-        )
-        self.register_buffer("imagenet_mean", torch.tensor([0.485, 0.456, 0.406]).view(1, 3, 1, 1), persistent=False)
-        self.register_buffer("imagenet_std", torch.tensor([0.229, 0.224, 0.225]).view(1, 3, 1, 1), persistent=False)
-
-    def _branch_features(self, x: torch.Tensor):
-        raw = x.clamp(0, 1)
-        vit_x = (raw - self.imagenet_mean) / self.imagenet_std
-        clip_x = (raw - self.clip_mean) / self.clip_std.clamp_min(1e-6)
-        vit_feat = self.vit_encoder.features(vit_x)
-        clip_feat = self.clip_encoder.features(clip_x)
-        fft_feat = self.fft_branch(raw)
-        srm_feat = self.srm_branch(raw)
-        return vit_feat, clip_feat, fft_feat, srm_feat
-
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        vit_feat, clip_feat, fft_feat, srm_feat = self._branch_features(x)
-        return self.head(torch.cat([vit_feat, clip_feat, fft_feat, srm_feat], dim=1))
-
-    def forward_breakdown(self, x: torch.Tensor):
-        """최종 로짓과 함께 각 관점별(구조/맥락/디테일) 실제 로짓을 반환합니다.
-
-        - structure: ViT 인코더의 자체 분류 헤드 (전역 구조/형태)
-        - context:  CLIP 인코더의 자체 분류 헤드 (의미적 맥락)
-        - detail:   주파수(FFT)+노이즈(SRM) 브랜치만 통과시킨 메인 헤드 (질감/흔적)
-
-        모두 학습된 가중치를 사용하는 실제 추론 결과이며, 임의 값이 아닙니다.
-        """
-        vit_feat, clip_feat, fft_feat, srm_feat = self._branch_features(x)
-        final_logits = self.head(torch.cat([vit_feat, clip_feat, fft_feat, srm_feat], dim=1))
-
-        structure_logits = self.vit_encoder.head(vit_feat)
-        context_logits = self.clip_encoder.head(clip_feat)
-        # 디테일 관점: 구조/맥락 특징을 0으로 두고 주파수·노이즈 특징만으로 판단
-        detail_logits = self.head(
-            torch.cat([torch.zeros_like(vit_feat), torch.zeros_like(clip_feat), fft_feat, srm_feat], dim=1)
-        )
-        return final_logits, {
-            "structure": structure_logits,
-            "context": context_logits,
-            "detail": detail_logits,
-        }
-
-
-def build_artifact_lora_hybrid() -> ArtifactLoRAHybrid:
-    return ArtifactLoRAHybrid()
-
-
-def is_artifact_lora_state_dict(state_dict: dict[str, Any]) -> bool:
-    return (
-        "vit_encoder.backbone.class_token" in state_dict
-        and "clip_encoder.vision_model.visual_projection.weight" in state_dict
-        and "fft_branch.proj.1.weight" in state_dict
-        and "srm_branch.highpass.weight" in state_dict
-        and "head.7.weight" in state_dict
-    )

inference.py

@@ -1,134 +0,0 @@
-import hashlib
-import io
-import random
-import time
-from typing import Any
-
-import numpy as np
-import torch
-from PIL import Image, UnidentifiedImageError
-
-from model_loader import LoadedModel, read_input_size, read_norm
-from xai import deterministic_heatmap, gradcam_or_fallback, overlay_heatmap, png_b64
-
-
-class InvalidImageError(ValueError):
-    pass
-
-
-def run_inference(image_bytes: bytes, loaded: LoadedModel) -> dict[str, Any]:
-    started = time.perf_counter()
-    original = _decode_image(image_bytes)
-
-    if loaded.is_mock:
-        score = _mock_score(image_bytes)
-        heatmap = deterministic_heatmap(score)
-        overlay = overlay_heatmap(original, heatmap)
-        return {
-            "score": score,
-            "verdict": "AI" if score >= 0.5 else "REAL",
-            "heatmap_b64": png_b64(heatmap),
-            "overlay_b64": png_b64(overlay),
-            "model_version": loaded.model_version,
-            "elapsed_ms": _elapsed_ms(started),
-        }
-
-    if loaded.load_error:
-        raise RuntimeError(f"Model not loaded: {loaded.load_error}")
-
-    input_tensor = _preprocess(original)
-    score, breakdown = _score_with_breakdown(loaded.model, input_tensor, image_bytes)
-    heatmap_b64, overlay_b64 = gradcam_or_fallback(loaded.model, input_tensor, original, score)
-    result = {
-        "score": score,
-        "verdict": "AI" if score >= 0.5 else "REAL",
-        "heatmap_b64": heatmap_b64,
-        "overlay_b64": overlay_b64,
-        "model_version": loaded.model_version,
-        "elapsed_ms": _elapsed_ms(started),
-    }
-    if breakdown is not None:
-        result["structure_score"] = breakdown["structure"]
-        result["context_score"] = breakdown["context"]
-        result["detail_score"] = breakdown["detail"]
-    return result
-
-
-def _decode_image(image_bytes: bytes) -> Image.Image:
-    try:
-        return Image.open(io.BytesIO(image_bytes)).convert("RGB")
-    except (UnidentifiedImageError, OSError) as exc:
-        raise InvalidImageError("Invalid image") from exc
-
-
-def _preprocess(image: Image.Image) -> torch.Tensor:
-    input_size = read_input_size()
-    resized = image.resize((input_size, input_size))
-    array = np.asarray(resized, dtype=np.float32) / 255.0
-    tensor = torch.from_numpy(array).permute(2, 0, 1).unsqueeze(0)
-    return tensor
-
-
-def _preprocess_imagenet(image: Image.Image) -> torch.Tensor:
-    input_size = read_input_size()
-    mean, std = read_norm()
-    resized = image.resize((input_size, input_size))
-    array = np.asarray(resized, dtype=np.float32) / 255.0
-    array = (array - np.array(mean, dtype=np.float32)) / np.array(std, dtype=np.float32)
-    tensor = torch.from_numpy(array).permute(2, 0, 1).unsqueeze(0)
-    return tensor
-
-
-def _logits_to_score(output: torch.Tensor) -> float:
-    flat = output.detach().float().reshape(output.shape[0], -1)
-    if flat.shape[1] == 1:
-        score = torch.sigmoid(flat[0, 0]).item()
-    else:
-        score = torch.softmax(flat[0, :2], dim=0)[1].item()
-    return float(max(0.0, min(1.0, score)))
-
-
-def _score_with_breakdown(
-    model: Any, input_tensor: torch.Tensor, image_bytes: bytes
-) -> tuple[float, dict[str, float] | None]:
-    """최종 점수와 관점별 실제 세부 점수(가능하면)를 함께 계산합니다."""
-    if hasattr(model, "forward_breakdown"):
-        try:
-            with torch.no_grad():
-                final_logits, parts = model.forward_breakdown(input_tensor)
-            score = _logits_to_score(final_logits)
-            breakdown = {key: _logits_to_score(value) for key, value in parts.items()}
-            return score, breakdown
-        except Exception:  # noqa: BLE001 - 세부 점수 실패가 전체 분석을 막지 않도록 함
-            pass
-    return _score_model(model, input_tensor, image_bytes), None
-
-
-def _score_model(model: Any, input_tensor: torch.Tensor, image_bytes: bytes) -> float:
-    if not hasattr(model, "__call__"):
-        return _mock_score(image_bytes)
-    if not getattr(model, "expects_raw_input", False):
-        input_tensor = _preprocess_imagenet(Image.open(io.BytesIO(image_bytes)).convert("RGB"))
-    with torch.no_grad():
-        output = model(input_tensor)
-    if isinstance(output, (tuple, list)):
-        output = output[0]
-    if not isinstance(output, torch.Tensor):
-        return _mock_score(image_bytes)
-    flat = output.detach().float().reshape(output.shape[0], -1)
-    if flat.shape[1] == 1:
-        score = torch.sigmoid(flat[0, 0]).item()
-    else:
-        score = torch.softmax(flat[0, :2], dim=0)[1].item()
-    return float(max(0.0, min(1.0, score)))
-
-
-def _mock_score(image_bytes: bytes) -> float:
-    digest = hashlib.sha256(image_bytes).digest()
-    seed = int.from_bytes(digest[:8], "big")
-    random.seed(seed)
-    return random.random()
-
-
-def _elapsed_ms(started: float) -> int:
-    return int((time.perf_counter() - started) * 1000)

main.py

@@ -1,67 +0,0 @@
-from contextlib import asynccontextmanager
-from typing import AsyncIterator
-
-from fastapi import FastAPI, File, HTTPException, UploadFile
-from fastapi.middleware.cors import CORSMiddleware
-
-from inference import InvalidImageError, run_inference
-from model_loader import LoadedModel, ensure_model_file, load_model, read_model_path
-from schemas import AnalyzeResponse, HealthResponse, VersionResponse
-
-MAX_IMAGE_BYTES = 10 * 1024 * 1024
-
-
-@asynccontextmanager
-async def lifespan(app: FastAPI) -> AsyncIterator[None]:
-    model_path = read_model_path()
-    ensure_model_file(model_path)
-    app.state.loaded_model = load_model(model_path)
-    yield
-
-
-app = FastAPI(title="A-EYE Inference API", version="1.0.0", lifespan=lifespan)
-
-# 모바일 앱은 쿠키/세션 자격증명을 사용하지 않으므로 credentials 를 비활성화합니다.
-# (allow_origins=["*"] 와 allow_credentials=True 조합은 브라우저에서 거부되며 보안상 위험합니다.)
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=False,
-    allow_methods=["GET", "POST", "OPTIONS"],
-    allow_headers=["*"],
-)
-
-
-@app.get("/api/health", response_model=HealthResponse)
-async def health() -> HealthResponse:
-    return HealthResponse(status="ok")
-
-
-@app.get("/api/version", response_model=VersionResponse)
-async def version() -> VersionResponse:
-    loaded = _loaded_model()
-    return VersionResponse(model_version=loaded.model_version, model_path=loaded.model_path)
-
-
-@app.post("/api/analyze", response_model=AnalyzeResponse)
-async def analyze(image: UploadFile = File(...)) -> AnalyzeResponse:
-    content_type = image.content_type or ""
-    if content_type not in {"image/jpeg", "image/png", "image/webp"}:
-        raise HTTPException(status_code=400, detail="Invalid image")
-    image_bytes = await image.read()
-    if not image_bytes or len(image_bytes) > MAX_IMAGE_BYTES:
-        raise HTTPException(status_code=400, detail="Invalid image")
-    try:
-        result = run_inference(image_bytes, _loaded_model())
-        return AnalyzeResponse(**result)
-    except InvalidImageError:
-        raise HTTPException(status_code=400, detail="Invalid image") from None
-    except RuntimeError as exc:
-        message = str(exc)
-        if message.startswith("Model not loaded"):
-            raise HTTPException(status_code=503, detail=message) from exc
-        raise
-
-
-def _loaded_model() -> LoadedModel:
-    return app.state.loaded_model

model_loader.py

@@ -1,117 +0,0 @@
-import logging
-import os
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Any, Optional
-
-import torch
-from dotenv import load_dotenv
-
-from artifact_model import build_artifact_lora_hybrid, is_artifact_lora_state_dict
-
-logger = logging.getLogger("aeye.model")
-load_dotenv()
-
-
-@dataclass(frozen=True)
-class LoadedModel:
-    model: Any
-    model_path: str
-    model_version: str
-    is_mock: bool
-    load_error: Optional[str] = None
-
-
-def read_model_path() -> str:
-    return os.getenv("MODEL_PATH", "./models/best.pt")
-
-
-def ensure_model_file(model_path: Optional[str] = None) -> None:
-    """모델 파일이 없고 MODEL_URL 이 설정돼 있으면 시작 시 한 번 내려받습니다.
-
-    737MB 모델 가중치는 git 저장소에 올리지 않으므로, 클라우드 배포 시
-    Hugging Face Hub / S3 / 직접 링크 등의 MODEL_URL 에서 받아오도록 합니다.
-    """
-    import urllib.request
-
-    path = Path(model_path or read_model_path())
-    if path.exists() and path.stat().st_size > 0:
-        return
-    url = os.getenv("MODEL_URL", "").strip()
-    if not url:
-        logger.warning("MODEL_URL 이 없고 %s 도 없습니다. MOCK 모드로 동작합니다.", path)
-        return
-    path.parent.mkdir(parents=True, exist_ok=True)
-    logger.info("모델 다운로드 시작: %s -> %s", url, path)
-    tmp = path.with_suffix(path.suffix + ".part")
-    try:
-        urllib.request.urlretrieve(url, tmp)  # noqa: S310 - 운영자가 지정한 신뢰된 URL
-        tmp.replace(path)
-        logger.info("모델 다운로드 완료: %s (%d bytes)", path, path.stat().st_size)
-    except Exception:
-        logger.exception("모델 다운로드 실패")
-        if tmp.exists():
-            tmp.unlink(missing_ok=True)
-
-
-def read_input_size() -> int:
-    raw = os.getenv("INPUT_SIZE", "224")
-    try:
-        return max(32, int(raw))
-    except ValueError:
-        return 224
-
-
-def read_norm() -> tuple[list[float], list[float]]:
-    mean = _read_float_list("IMAGE_MEAN", [0.485, 0.456, 0.406])
-    std = _read_float_list("IMAGE_STD", [0.229, 0.224, 0.225])
-    return mean, std
-
-
-def load_model(model_path: Optional[str] = None) -> LoadedModel:
-    path = Path(model_path or read_model_path())
-    if not path.exists() or path.stat().st_size == 0:
-        logger.warning("MOCK MODE: %s not found, using random predictions", path)
-        return LoadedModel(model=None, model_path="MOCK", model_version="mock-0.0.0", is_mock=True)
-
-    version = _read_version(path)
-    try:
-        checkpoint = torch.load(path, map_location="cpu")
-        if isinstance(checkpoint, dict) and "model_state_dict" in checkpoint:
-            state_dict = checkpoint["model_state_dict"]
-            if isinstance(state_dict, dict) and is_artifact_lora_state_dict(state_dict):
-                model = build_artifact_lora_hybrid()
-                missing, unexpected = model.load_state_dict(state_dict, strict=False)
-                if missing or unexpected:
-                    logger.warning("Artifact model loaded with missing=%s unexpected=%s", missing, unexpected)
-            else:
-                raise RuntimeError("Unsupported checkpoint state_dict architecture")
-        else:
-            model = checkpoint
-        if hasattr(model, "eval"):
-            model.eval()
-        logger.info("Loaded model from %s version %s", path, version)
-        return LoadedModel(model=model, model_path=str(path), model_version=version, is_mock=False)
-    except Exception as exc:
-        reason = f"{type(exc).__name__}: {exc}"
-        logger.exception("Model load failed")
-        return LoadedModel(model=None, model_path=str(path), model_version=version, is_mock=False, load_error=reason)
-
-
-def _read_version(path: Path) -> str:
-    sidecar = Path(f"{path}.version.txt")
-    if not sidecar.exists():
-        return "unknown"
-    value = sidecar.read_text(encoding="utf-8").strip()
-    return value or "unknown"
-
-
-def _read_float_list(key: str, fallback: list[float]) -> list[float]:
-    raw = os.getenv(key)
-    if not raw:
-        return fallback
-    try:
-        values = [float(part.strip()) for part in raw.split(",")]
-        return values if len(values) == 3 else fallback
-    except ValueError:
-        return fallback

models/best.pt.version.txt

@@ -1 +0,0 @@
-24_realworld_artifact_30ep

models/best_model.pt.version.txt

@@ -1 +0,0 @@
-24_realworld_artifact_30ep

requirements.txt

@@ -2,7 +2,5 @@ flask==3.0.3
 pyyaml>=6.0
 pillow>=10.0.0
 numpy>=1.24.0
-torch>=2.1.0
-torchvision>=0.16.0
 transformers>=4.38.0
 huggingface_hub>=0.24.0

schemas.py

@@ -1,23 +0,0 @@
-from pydantic import BaseModel, Field
-
-
-class AnalyzeResponse(BaseModel):
-    score: float = Field(ge=0.0, le=1.0)
-    verdict: str
-    heatmap_b64: str
-    overlay_b64: str
-    model_version: str
-    elapsed_ms: int
-    # 관점별 실제 세부 점수(AI 확률). 모델이 지원할 때만 채워집니다.
-    structure_score: float | None = Field(default=None, ge=0.0, le=1.0)
-    context_score: float | None = Field(default=None, ge=0.0, le=1.0)
-    detail_score: float | None = Field(default=None, ge=0.0, le=1.0)
-
-
-class VersionResponse(BaseModel):
-    model_version: str
-    model_path: str
-
-
-class HealthResponse(BaseModel):
-    status: str

xai.py

@@ -1,157 +0,0 @@
-import base64
-import io
-from typing import Any, Optional
-
-import numpy as np
-import torch
-from PIL import Image
-
-CAM_SIZE = 224
-
-
-def png_b64(image: Image.Image) -> str:
-    buffer = io.BytesIO()
-    image.save(buffer, format="PNG")
-    return base64.b64encode(buffer.getvalue()).decode("ascii")
-
-
-# ─── 합성(폴백) 히트맵 ────────────────────────────────────────────────
-# GradCAM 이 불가능한 경우(모델 미로드/예외)에만 사용하는 점수 기반 대체 시각화.
-
-def deterministic_heatmap(score: float, size: int = CAM_SIZE) -> Image.Image:
-    x = np.linspace(0, 1, size, dtype=np.float32)
-    y = np.linspace(0, 1, size, dtype=np.float32)
-    xx, yy = np.meshgrid(x, y)
-    center_x = 0.28 + (score * 0.44)
-    center_y = 0.72 - (score * 0.38)
-    primary = np.exp(-(((xx - center_x) ** 2) / 0.055 + ((yy - center_y) ** 2) / 0.085))
-    secondary = np.exp(-(((xx - 0.72) ** 2) / 0.09 + ((yy - 0.28) ** 2) / 0.06)) * (0.35 + score * 0.35)
-    heat = np.clip(primary + secondary, 0, 1)
-    rgba = np.zeros((size, size, 4), dtype=np.uint8)
-    rgba[..., 0] = (255 * heat).astype(np.uint8)
-    rgba[..., 1] = (180 * (1 - heat)).astype(np.uint8)
-    rgba[..., 2] = np.uint8(255 * (0.25 + score * 0.25))
-    rgba[..., 3] = 220
-    return Image.fromarray(rgba, "RGBA")
-
-
-def overlay_heatmap(original: Image.Image, heatmap: Image.Image, alpha: float = 0.5) -> Image.Image:
-    base = original.convert("RGBA")
-    heat = heatmap.convert("RGBA").resize(base.size)
-    return Image.blend(base, heat, alpha).convert("RGB")
-
-
-# ─── ViT/CLIP 토큰 출력을 공간 특징맵으로 변환 ────────────────────────
-# 트랜스포머 레이어 출력은 [B, tokens, C] 형태이므로 CLS 토큰을 제거하고
-# [B, C, H, W] 격자로 reshape 해야 GradCAM 이 공간 히트맵을 만들 수 있습니다.
-
-def _vit_reshape_transform(tensor: torch.Tensor) -> torch.Tensor:
-    tokens = tensor[:, 1:, :]
-    b, n, c = tokens.shape
-    side = int(round(n ** 0.5))
-    tokens = tokens.reshape(b, side, side, c)
-    return tokens.permute(0, 3, 1, 2)
-
-
-def _resolve_attr(model: Any, path: str) -> Optional[Any]:
-    obj: Any = model
-    for part in path.split("."):
-        if part.endswith("]"):
-            name, idx = part[:-1].split("[")
-            obj = getattr(obj, name, None)
-            if obj is None:
-                return None
-            try:
-                obj = obj[int(idx)]
-            except (IndexError, ValueError, TypeError):
-                return None
-        else:
-            obj = getattr(obj, part, None)
-        if obj is None:
-            return None
-    return obj
-
-
-def find_target_layer(model: Any) -> tuple[Optional[Any], bool]:
-    """(target_layer, needs_reshape) 를 반환합니다.
-
-    A-EYE 하이브리드 모델에서 의심 영역 시각화에 가장 적합한 타겟은
-    공간 정보를 유지하며 최종 판단에 기여하는 컨볼루션 브랜치입니다.
-
-    - srm_branch(노이즈 잔차): 이미지 공간의 위·변조 흔적에 반응 → 1순위
-    - fft_branch(주파수): 생성 모델 특유의 주파수 패턴 → 2순위
-    - clip_encoder fc2: 트랜스포머(요청 시), CLS-only 풀링이면 신호가 약함 → 폴백
-
-    (CLIP/ViT 브랜치는 CLS 토큰만 출력에 사용하므로 패치 토큰 기반
-     GradCAM 이 사실상 무의미해, 공간 컨볼루션 브랜치를 우선합니다.)
-    """
-    srm = _resolve_attr(model, "srm_branch.encoder")
-    if srm is not None:
-        return srm, False
-
-    fft = _resolve_attr(model, "fft_branch.conv")
-    if fft is not None:
-        return fft, False
-
-    clip_fc2 = _resolve_attr(
-        model, "clip_encoder.vision_model.vision_model.encoder.layers[-1].mlp.fc2"
-    )
-    if clip_fc2 is not None:
-        return clip_fc2, True
-
-    if not hasattr(model, "named_modules"):
-        return None, False
-    modules = list(model.named_modules())
-    for name, module in reversed(modules):
-        lowered = name.lower()
-        if any(token in lowered for token in ("layer4", "conv", "blocks")):
-            return module, False
-    return (modules[-1][1] if modules else None), False
-
-
-def _colorize_cam(grayscale: np.ndarray) -> np.ndarray:
-    """0~1 CAM 을 JET 컬러맵 RGB 이미지(uint8)로 변환."""
-    gray_u8 = np.uint8(255 * np.clip(grayscale, 0, 1))
-    try:
-        import cv2
-
-        colored = cv2.applyColorMap(gray_u8, cv2.COLORMAP_JET)
-        return cv2.cvtColor(colored, cv2.COLOR_BGR2RGB)
-    except Exception:
-        rgb = np.zeros((*gray_u8.shape, 3), dtype=np.uint8)
-        rgb[..., 0] = gray_u8
-        rgb[..., 2] = 255 - gray_u8
-        return rgb
-
-
-def gradcam_or_fallback(
-    model: Any, input_tensor: torch.Tensor, original: Image.Image, score: float
-) -> tuple[str, str]:
-    try:
-        from pytorch_grad_cam import GradCAM
-        from pytorch_grad_cam.utils.image import show_cam_on_image
-        from pytorch_grad_cam.utils.model_targets import ClassifierOutputTarget
-
-        target_layer, needs_reshape = find_target_layer(model)
-        if target_layer is None:
-            raise RuntimeError("No target layer")
-
-        reshape = _vit_reshape_transform if needs_reshape else None
-        target_idx = 1 if score >= 0.5 else 0
-
-        with GradCAM(model=model, target_layers=[target_layer], reshape_transform=reshape) as cam:
-            grayscale = cam(input_tensor=input_tensor, targets=[ClassifierOutputTarget(target_idx)])[0]
-
-        grayscale = np.maximum(grayscale, 0)
-        grayscale = grayscale / (grayscale.max() + 1e-8)
-
-        rgb = np.asarray(original.resize((CAM_SIZE, CAM_SIZE)).convert("RGB"), dtype=np.float32) / 255.0
-        overlay = show_cam_on_image(rgb, grayscale, use_rgb=True, image_weight=0.55)
-        overlay_image = Image.fromarray(np.ascontiguousarray(overlay, dtype=np.uint8))
-
-        heatmap_image = Image.fromarray(_colorize_cam(grayscale))
-        return png_b64(heatmap_image), png_b64(overlay_image)
-    except Exception:
-        heatmap = deterministic_heatmap(score)
-        overlay = overlay_heatmap(original, heatmap)
-        return png_b64(heatmap), png_b64(overlay)