File size: 3,476 Bytes
0ef5f4e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | 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) |