File size: 3,138 Bytes
268f99f |
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 |
import torch
import torch.nn as nn
from torchvision import transforms
from PIL import Image
import io
import base64
from typing import Dict, List, Any
import timm
class EndpointHandler:
def __init__(self, path=""):
"""
Initialize handler with model path
"""
self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# Class names
self.class_names = ['Invalid', 'SDTI', 'Stage_I', 'Stage_II', 'Stage_III', 'Stage_IV', 'Unstageable']
# Load RexNet model
self.model = timm.create_model('rexnet_150', pretrained=False, num_classes=7)
# Load state dict
model_path = f"{path}/pytorch_model.bin" if path else "pytorch_model.bin"
state_dict = torch.load(model_path, map_location=self.device)
self.model.load_state_dict(state_dict)
self.model.to(self.device)
self.model.eval()
# Define preprocessing
self.transform = transforms.Compose([
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])
def __call__(self, data: Dict[str, Any]) -> List[Dict[str, Any]]:
"""
Process inference request
"""
# Get inputs
inputs = data.pop("inputs", data)
# Handle different input formats
if isinstance(inputs, dict) and "image" in inputs:
image_data = inputs["image"]
elif isinstance(inputs, str):
image_data = inputs
else:
raise ValueError("Invalid input format. Expected {'image': base64_string} or base64_string")
# Decode base64 image
try:
image_bytes = base64.b64decode(image_data)
image = Image.open(io.BytesIO(image_bytes)).convert('RGB')
except Exception as e:
raise ValueError(f"Failed to decode image: {str(e)}")
# Preprocess image
image_tensor = self.transform(image).unsqueeze(0).to(self.device)
# Run inference
with torch.no_grad():
outputs = self.model(image_tensor)
probabilities = torch.nn.functional.softmax(outputs, dim=1)
# Get top 3 predictions
top3_prob, top3_indices = torch.topk(probabilities, 3)
# Prepare response
predictions = []
for i in range(3):
predictions.append({
"label": self.class_names[top3_indices[0][i].item()],
"score": float(top3_prob[0][i].item())
})
# Get all probabilities
all_probs = {}
for i, class_name in enumerate(self.class_names):
all_probs[class_name] = float(probabilities[0][i].item())
return [{
"predictions": predictions,
"probabilities": all_probs,
"predicted_class": predictions[0]["label"],
"confidence": predictions[0]["score"]
}] |