Create handler.py

handler.py

@@ -0,0 +1,56 @@
+import base64
+import io
+from typing import Any, Dict
+
+import torch
+from PIL import Image
+from transformers import CLIPModel, CLIPProcessor
+
+class EndpointHandler:
+    def __init__(self, model_dir: str = "", **kwargs: Any):
+        self.device = "cuda" if torch.cuda.is_available() else "cpu"
+        self.model = CLIPModel.from_pretrained(model_dir).to(self.device)
+        self.processor = CLIPProcessor.from_pretrained(model_dir)
+        self.model.eval()
+
+    def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]:
+        # Flexibel: data kan direct inputs bevatten of via "inputs"
+        payload = data.get("inputs", data)
+
+        text = payload.get("text")
+        image_input = payload.get("image")  # base64 string of URL
+
+        # Image verwerken (base64 of PIL)
+        image = None
+        if isinstance(image_input, str):
+            if image_input.startswith(("http://", "https://")):
+                # URL-support (optioneel, vereist requests + PIL)
+                import requests
+                response = requests.get(image_input)
+                image = Image.open(io.BytesIO(response.content)).convert("RGB")
+            else:
+                # base64
+                image_bytes = base64.b64decode(image_input.split(",")[-1])
+                image = Image.open(io.BytesIO(image_bytes)).convert("RGB")
+
+        with torch.no_grad():
+            if image is not None and text is not None:
+                # Beide → image + text embeddings
+                inputs = self.processor(text=text, images=image, return_tensors="pt", padding=True).to(self.device)
+                outputs = self.model(**inputs)
+                return {
+                    "image_embedding": outputs.image_embeds[0].cpu().tolist(),
+                    "text_embedding": outputs.text_embeds[0].cpu().tolist()
+                }
+            elif image is not None:
+                # Alleen image
+                inputs = self.processor(images=image, return_tensors="pt").to(self.device)
+                image_features = self.model.get_image_features(**inputs)
+                return {"image_embedding": image_features[0].cpu().tolist()}
+            elif text is not None:
+                # Alleen text
+                inputs = self.processor(text=text, return_tensors="pt", padding=True).to(self.device)
+                text_features = self.model.get_text_features(**inputs)
+                return {"text_embedding": text_features[0].cpu().tolist()}
+            else:
+                return {"error": "Geef 'text' of 'image' (base64) mee"}