handler.py

from typing import Any, Dict, List
import base64, io, os
from PIL import Image
import torch

from transformers import AutoProcessor
# Use the vendor'ed GUI-Actor sources (copied into the repo as /gui_actor)
from gui_actor.modeling_qwen25vl import Qwen2_5_VLForConditionalGenerationWithPointer
from gui_actor.inference import inference

class EndpointHandler:
    """
    Accepts JSON like:
    {
      "image": "data:image/png;base64,...."  OR  "image_b64": "<raw_base64>",
      "image_url": "https://...png",         # optional (qwen-vl-utils supports it)
      "prompt": "Click the close button",
      "topk": 3,
      "return_pixels": true,
      "screen_w": 1920,
      "screen_h": 1080
    }
    Returns:
    {
      "points_norm": [[x,y], ...],   # 0..1 normalized
      "points_px":   [[x_px,y_px], ...]  # if screen_w/h given
    }
    """

    def __init__(self, path: str = ""):
        device = "cuda" if torch.cuda.is_available() else "cpu"
        # bfloat16 on GPU is fine for Qwen2.5; fallback to float32 on CPU
        dtype = torch.bfloat16 if device == "cuda" else torch.float32

        self.processor = AutoProcessor.from_pretrained(path)
        self.tokenizer = self.processor.tokenizer

        # Avoid hard requiring flash-attn; it will use PyTorch SDPA if unavailable
        self.model = Qwen2_5_VLForConditionalGenerationWithPointer.from_pretrained(
            path, torch_dtype=dtype, device_map="auto"
        ).eval()

    def _load_pil(self, data: Dict[str, Any]) -> Image.Image:
        if "image" in data and isinstance(data["image"], str) and data["image"].startswith("data:image"):
            # "data:image/png;base64,......"
            b64 = data["image"].split("base64,")[-1]
            return Image.open(io.BytesIO(base64.b64decode(b64))).convert("RGB")
        if "image_b64" in data:
            return Image.open(io.BytesIO(base64.b64decode(data["image_b64"]))).convert("RGB")
        # If image_url is provided, pass URL through; Qwen utils can handle URLs internally
        return None

    def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]:
        # Handle both direct input and HuggingFace's nested input format
        if "inputs" in data:
            payload = data["inputs"]
        else:
            payload = data
        
        prompt = payload.get("prompt", "")
        topk = int(payload.get("topk", 3))
        img = self._load_pil(payload)

        # Build conversation per model card
        user_content = []
        if img is not None:
            user_content.append({"type": "image", "image": img})
        elif "image_url" in payload:
            user_content.append({"type": "image", "image_url": payload["image_url"]})
        else:
            raise ValueError("No image provided. Supply 'image' (data URL), 'image_b64', or 'image_url'.")

        user_content.append({"type": "text", "text": prompt})
        conversation = [
            {
                "role": "system",
                "content": [{"type": "text",
                             "text": "You are a GUI agent. Given a screenshot of the current GUI and a human instruction, "
                                     "locate the target element and output a click position."}]
            },
            {"role": "user", "content": user_content},
        ]

        try:
            pred = inference(
                conversation,
                self.model,
                self.tokenizer,
                self.processor,
                use_placeholder=True,
                topk=topk,
            )

            points = pred.get("topk_points") or []
            result = {"points_norm": points}

            # Optional: convert to pixels
            if payload.get("return_pixels") and payload.get("screen_w") and payload.get("screen_h"):
                w = int(payload["screen_w"])
                h = int(payload["screen_h"])
                result["points_px"] = [[int(x*w), int(y*h)] for (x, y) in points]

            return result
            
        except Exception as e:
            return {"error": str(e), "points_norm": [], "points_px": []}