handler.py

import torch
import base64
import io
from typing import Dict, List, Any
from PIL import Image
from torchvision import transforms, models
import torch.nn as nn
import torch.nn.functional as F


class ResNet50Classifier(nn.Module):
    def __init__(self, num_classes: int = 3, dropout: float = 0.3):
        super().__init__()
        backbone = models.resnet50(weights=None)
        self.features = nn.Sequential(*list(backbone.children())[:-1])
        self.classifier = nn.Sequential(
            nn.Linear(2048, 512), nn.GELU(), nn.Dropout(dropout),
            nn.Linear(512, 128),  nn.GELU(), nn.Dropout(dropout),
            nn.Linear(128, num_classes),
        )

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.classifier(self.features(x).flatten(1))


class EndpointHandler:
    CLASSES = ["benign", "malignant", "normal"]

    # Identical to the val/test transform in the notebook
    TRANSFORM = transforms.Compose([
        transforms.Grayscale(num_output_channels=3),
        transforms.Resize((224, 224)),
        transforms.ToTensor(),
        transforms.Normalize(
            mean=[0.485, 0.456, 0.406],
            std= [0.229, 0.224, 0.225],
        ),
    ])

    def __init__(self, path: str = ""):
        """
        Loads the ResNet-50 mammography classifier from a local checkpoint.

        Args:
            path: Directory containing best_model.pth  (HF Inference Endpoints
                  sets this to the local snapshot of your model repo).
        """
        self.device = "cuda" if torch.cuda.is_available() else "cpu"

        self.model = ResNet50Classifier(num_classes=len(self.CLASSES)).to(self.device)

        import os
        ckpt_path = os.path.join(path, "best_model.pth")
        self.model.load_state_dict(
            torch.load(ckpt_path, map_location=self.device)
        )
        self.model.eval()

    def __call__(self, data: Dict[str, Any]) -> List[Dict[str, Any]]:
        """
        Args:
            data: Dictionary with key "inputs" containing:
                  - "image": Base64-encoded PNG/JPG string

        Returns:
            List of {"label": str, "score": float} sorted by score descending.
            Example:
                [
                    {"label": "malignant", "score": 0.821},
                    {"label": "benign",    "score": 0.134},
                    {"label": "normal",    "score": 0.045},
                ]
        """
        inputs = data.pop("inputs", data)
        image_base64 = inputs.get("image")

        if not image_base64:
            return [{"error": "Missing 'image' in payload"}]

        # 1. Decode base64 image
        try:
            image = Image.open(
                io.BytesIO(base64.b64decode(image_base64))
            ).convert("RGB")
        except Exception as e:
            return [{"error": f"Failed to decode image: {str(e)}"}]

        # 2. Preprocess — same pipeline as notebook val/test transform
        tensor = self.TRANSFORM(image).unsqueeze(0).to(self.device)  # (1, 3, 224, 224)

        # 3. Inference
        with torch.no_grad():
            logits = self.model(tensor)                        # (1, 3)
            probs  = F.softmax(logits, dim=1).cpu().numpy()[0] # (3,)

        # 4. Format and sort
        results = [
            {"label": label, "score": float(prob)}
            for label, prob in zip(self.CLASSES, probs)
        ]

        return sorted(results, key=lambda x: x["score"], reverse=True)