Update handler.py

handler.py

@@ -1,91 +1,49 @@
-# hf_client.py
-from __future__ import annotations
-import os, json, base64, io, time
-from typing import Any, Dict, Optional, Union
-import requests
+# /repository/handler.py
+import base64, io, os, json
+from typing import Any, Dict, List
 from PIL import Image
 
-DEFAULT_TIMEOUT = 60
-
-class HFEndpointError(RuntimeError):
-    pass
-
-class HFECGClient:
-    """
-    Hugging Face Inference Endpoint (LLaVA/PULSE tarzı) istemci.
-    image: URL veya yerel dosya yolu (yerel dosyayı otomatik base64'e çevirir).
-    dönüş: {"text": "...", "raw": <json>, "latency_ms": float, "status_code": int}
-    """
-    def __init__(self, endpoint_url: str, hf_token: str, timeout: int = DEFAULT_TIMEOUT,
-                 session: Optional[requests.Session] = None) -> None:
-        if not endpoint_url: raise ValueError("endpoint_url zorunlu.")
-        if not hf_token: raise ValueError("hf_token zorunlu.")
-        self.endpoint_url = endpoint_url.strip()
-        self.hf_token = hf_token.strip()
-        self.timeout = timeout
-        self.session = session or requests.Session()
-
-    @staticmethod
-    def _encode_image_to_b64(img: Image.Image) -> str:
-        buf = io.BytesIO(); img.save(buf, format="PNG")
-        return "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode("utf-8")
-
-    def _image_to_payload(self, image: Union[str, Image.Image]) -> str:
-        if isinstance(image, str):
-            low = image.lower()
-            if low.startswith("http://") or low.startswith("https://"):
-                return image  # URL ise aynen gönder
-            if not os.path.isfile(image):
-                raise FileNotFoundError(f"Görsel bulunamadı: {image}")
-            return self._encode_image_to_b64(Image.open(image).convert("RGB"))
-        elif isinstance(image, Image.Image):
-            return self._encode_image_to_b64(image.convert("RGB"))
+# (Gerekiyorsa: from transformers import ...  # model yükleme burada olur)
+
+class EndpointHandler:
+    def __init__(self, path: str = "") -> None:
+        # Burada model/processor/tokenizer'ı yükleyin
+        # ör: self.model = ...
+        #      self.processor = ...
+        pass
+
+    def _load_image(self, img_field: str) -> Image.Image:
+        if img_field.startswith("data:image"):
+            # data URL -> bytes
+            header, b64data = img_field.split(",", 1)
+            img_bytes = base64.b64decode(b64data)
+            return Image.open(io.BytesIO(img_bytes)).convert("RGB")
+        elif img_field.startswith("http://") or img_field.startswith("https://"):
+            import requests
+            resp = requests.get(img_field, timeout=20)
+            resp.raise_for_status()
+            return Image.open(io.BytesIO(resp.content)).convert("RGB")
         else:
-            raise TypeError("image: str (url/dosya) veya PIL.Image olmalı.")
-
-    def build_payload(self, query: str, image: Union[str, Image.Image],
-                      max_new_tokens: int = 4096, temperature: float = 0.0, top_p: float = 1.0,
-                      repetition_penalty: float = 1.0, return_full_text: bool = False, seed: int = 42,
-                      conv_mode: str = "llava_v2", use_cache: bool = True,
-                      extra_parameters: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
-        payload = {
-            "inputs": {"query": query, "image": self._image_to_payload(image)},
-            "parameters": {
-                "max_new_tokens": max_new_tokens,
-                "do_sample": False if temperature == 0 else True,
-                "temperature": temperature, "top_p": top_p,
-                "repetition_penalty": repetition_penalty,
-                "return_full_text": return_full_text, "seed": seed,
-            },
-            "options": {"use_cache": use_cache},
-            "conv_mode": conv_mode,
-        }
-        if extra_parameters: payload["parameters"].update(extra_parameters)
-        return payload
-
-    def infer(self, query: str, image: Union[str, Image.Image], **gen_kwargs) -> Dict[str, Any]:
-        payload = self.build_payload(query, image, **gen_kwargs)
-        headers = {
-            "Accept": "application/json",
-            "Authorization": f"Bearer {self.hf_token}",
-            "Content-Type": "application/json",
-        }
-        t0 = time.time()
-        resp = self.session.post(self.endpoint_url, headers=headers, json=payload, timeout=self.timeout)
-        latency_ms = (time.time() - t0) * 1000.0
-        status = resp.status_code
-        try:
-            data = resp.json()
-        except Exception as e:
-            raise HFEndpointError(f"JSON olmayan cevap (status={status}): {resp.text[:500]}") from e
-
-        text = ""
-        if isinstance(data, list) and data:
-            text = data[0].get("generated_text", "") or data[0].get("text", "")
-        elif isinstance(data, dict):
-            text = data.get("generated_text", "") or data.get("text", "")
-
-        if status >= 400:
-            raise HFEndpointError(f"Endpoint hatası {status}: {json.dumps(data)[:500]}")
-
-        return {"text": text, "raw": data, "latency_ms": latency_ms, "status_code": status}
+            # Yerel yol (container içinden)
+            return Image.open(img_field).convert("RGB")
+
+    def __call__(self, data: Dict[str, Any]) -> List[Dict[str, Any]]:
+        """
+        Hugging Face Inference Toolkit burayı çağırır.
+        Beklenen dönüş genelde: [{"generated_text": "..."}]
+        """
+        inputs = data.get("inputs") or {}
+        params = data.get("parameters") or {}
+        query = inputs.get("query", "")
+        img_field = inputs.get("image", "")
+
+        # Görseli hazırla (opsiyonel — modeliniz görsel kullanıyorsa)
+        image = None
+        if img_field:
+            image = self._load_image(img_field)
+
+        # Burada kendi inference kodunuzu çağırın:
+        # out_text = run_model(self.model, self.processor, query, image, **params)
+        out_text = f"(demo) prompt='{query[:50]}...' image={'yes' if image else 'no'}"
+
+        return [{"generated_text": out_text}]