Delete mm_utils_local.py

mm_utils_local.py

@@ -1,259 +0,0 @@
-# mm_utils_local.py
-# LLaVA/PULSE uyumlu, dayanıklı mm_utils (anyres + pad)
-# - crop_size/size alanlarını güvenli okur
-# - preprocess veya __call__ farkını soğurur
-# - patch_size'a tam bölünecek pad ekler
-# - upstream imzalarıyla uyumludur
-
-from typing import Any, Dict, List, Optional, Sequence, Tuple
-from io import BytesIO
-import base64
-import math
-import ast
-
-import torch
-from PIL import Image
-from transformers import StoppingCriteria
-from llava.constants import IMAGE_TOKEN_INDEX  # imza uyumu için
-
-# ---------- Yardımcılar ----------
-
-def _get_crop_size(processor: Any, default: int = 224) -> int:
-    cs = getattr(processor, "crop_size", None)
-    if cs is None:
-        sz = getattr(processor, "size", None)
-        if isinstance(sz, dict):
-            return int(sz.get("shortest_edge", default))
-        if isinstance(sz, int):
-            return int(sz)
-        return int(default)
-    if isinstance(cs, dict):
-        if "height" in cs:
-            return int(cs["height"])
-        if "shortest_edge" in cs:
-            return int(cs["shortest_edge"])
-        # beklenmedik dict: ilk değeri al
-        for v in cs.values():
-            return int(v)
-    return int(cs)
-
-def _get_shortest_edge(processor: Any, fallback: Optional[int] = None) -> int:
-    sz = getattr(processor, "size", None)
-    if isinstance(sz, dict) and "shortest_edge" in sz:
-        return int(sz["shortest_edge"])
-    if isinstance(sz, int):
-        return int(sz)
-    return _get_crop_size(processor, default=(fallback or 224))
-
-def _preprocess_one(processor: Any, img: Image.Image) -> torch.Tensor:
-    # Bazı sürümlerde .preprocess yok; direkt __call__ çalıştırılır.
-    if hasattr(processor, "preprocess"):
-        out = processor.preprocess(img, return_tensors="pt")
-    else:
-        out = processor(img, return_tensors="pt")
-    return out["pixel_values"][0]
-
-def pad_to_multiple(image: Image.Image, multiple: int) -> Image.Image:
-    w, h = image.size
-    W = math.ceil(w / multiple) * multiple
-    H = math.ceil(h / multiple) * multiple
-    if (W, H) == (w, h):
-        return image
-    canvas = Image.new(image.mode, (W, H), (0, 0, 0))
-    canvas.paste(image, (0, 0))
-    return canvas
-
-# ---------- Orijinal API ----------
-
-def select_best_resolution(original_size: Tuple[int, int], possible_resolutions: List[Tuple[int, int]]) -> Tuple[int, int]:
-    """Upstream ile aynı mantık: en etkili ve en az boşa giden çözünürlüğü seç."""
-    original_width, original_height = original_size
-    best_fit = None
-    max_effective_resolution = 0
-    min_wasted_resolution = float("inf")
-    for width, height in possible_resolutions:
-        scale = min(width / original_width, height / original_height)
-        down_w, down_h = int(original_width * scale), int(original_height * scale)
-        effective = min(down_w * down_h, original_width * original_height)
-        wasted = (width * height) - effective
-        if (effective > max_effective_resolution) or (effective == max_effective_resolution and wasted < min_wasted_resolution):
-            max_effective_resolution = effective
-            min_wasted_resolution = wasted
-            best_fit = (width, height)
-    return best_fit
-
-def resize_and_pad_image(image: Image.Image, target_resolution: Tuple[int, int]) -> Image.Image:
-    """Hedef çözünürlüğe orantıyı koruyarak resize + siyah pad."""
-    ow, oh = image.size
-    W, H = target_resolution
-    sw, sh = W / ow, H / oh
-    if sw < sh:
-        nw, nh = W, min(math.ceil(oh * sw), H)
-    else:
-        nh, nw = H, min(math.ceil(ow * sh), W)
-    resized = image.resize((nw, nh))
-    canvas = Image.new("RGB", (W, H), (0, 0, 0))
-    canvas.paste(resized, ((W - nw) // 2, (H - nh) // 2))
-    return canvas
-
-def divide_to_patches(image: Image.Image, patch_size: int) -> List[Image.Image]:
-    """Görüntüyü patch_size x patch_size karelere böl."""
-    patches: List[Image.Image] = []
-    W, H = image.size
-    for y in range(0, H, patch_size):
-        for x in range(0, W, patch_size):
-            patches.append(image.crop((x, y, x + patch_size, y + patch_size)))
-    return patches
-
-def get_anyres_image_grid_shape(image_size: Tuple[int, int], grid_pinpoints, patch_size: int) -> Tuple[int, int]:
-    """AnyRes sonrası patch ızgara boyutu (W//patch, H//patch)."""
-    if isinstance(grid_pinpoints, list):
-        possible_resolutions = grid_pinpoints
-    else:
-        possible_resolutions = ast.literal_eval(grid_pinpoints)
-    width, height = select_best_resolution(image_size, possible_resolutions)
-    return width // patch_size, height // patch_size
-
-def process_anyres_image(image: Image.Image, processor: Any, grid_pinpoints) -> torch.Tensor:
-    """
-    Robust AnyRes:
-      - crop_size/size güvenli okuma
-      - hedef çözünürlüğe resize+pad
-      - patch_size'a tam bölünecek pad
-      - preprocess/call farkını soyutlama
-    """
-    if isinstance(grid_pinpoints, list):
-        possible_resolutions = grid_pinpoints
-    else:
-        possible_resolutions = ast.literal_eval(grid_pinpoints)
-
-    patch_size = _get_crop_size(processor, default=224)
-    shortest_edge = _get_shortest_edge(processor, fallback=patch_size)
-
-    best_resolution = select_best_resolution(image.size, possible_resolutions)
-    image_padded = resize_and_pad_image(image, best_resolution)
-    image_padded = pad_to_multiple(image_padded, patch_size)
-
-    patches = divide_to_patches(image_padded, patch_size)
-    image_original_resize = image.resize((shortest_edge, shortest_edge))
-
-    image_patches = [_preprocess_one(processor, image_original_resize)]
-    image_patches += [_preprocess_one(processor, p) for p in patches]
-    return torch.stack(image_patches, dim=0)
-
-def load_image_from_base64(image: str) -> Image.Image:
-    return Image.open(BytesIO(base64.b64decode(image)))
-
-def expand2square(pil_img: Image.Image, background_color: Tuple[int, int, int]) -> Image.Image:
-    w, h = pil_img.size
-    if w == h:
-        return pil_img
-    if w > h:
-        result = Image.new(pil_img.mode, (w, w), background_color)
-        result.paste(pil_img, (0, (w - h) // 2))
-        return result
-    result = Image.new(pil_img.mode, (h, h), background_color)
-    result.paste(pil_img, ((h - w) // 2, 0))
-    return result
-
-def process_images(images: List[Image.Image], image_processor: Any, model_cfg: Any):
-    """
-    Upstream API ile aynı isim/geri dönüş; ancak daha dayanıklı:
-      - pad: image_mean yoksa güvenli varsayılan (0.5,0.5,0.5)
-      - anyres: robust process_anyres_image
-      - else: toplu çağrı TypeError ise tek tek çağrı fallback
-    """
-    # bazı konfig’lerde alan adı mm_image_aspect_ratio olabilir
-    image_aspect_ratio = getattr(model_cfg, "image_aspect_ratio", None) or getattr(model_cfg, "mm_image_aspect_ratio", None)
-    new_images: List[torch.Tensor] = []
-
-    if image_aspect_ratio == "pad":
-        for image in images:
-            img_mean = getattr(image_processor, "image_mean", [0.5, 0.5, 0.5])
-            bg = tuple(int(x * 255) for x in img_mean)
-            image_sq = expand2square(image, bg)
-            image_t = _preprocess_one(image_processor, image_sq)
-            new_images.append(image_t)
-
-    elif image_aspect_ratio == "anyres":
-        grid = getattr(model_cfg, "image_grid_pinpoints", "[(336,336)]")
-        for image in images:
-            image_t = process_anyres_image(image, image_processor, grid)
-            new_images.append(image_t)
-
-    else:
-        try:
-            out = image_processor(images, return_tensors="pt")
-            return out["pixel_values"]
-        except TypeError:
-            outs = [image_processor(img, return_tensors="pt") for img in images]
-            pix = [o["pixel_values"][0] for o in outs]
-            return torch.stack(pix, dim=0)
-
-    if all(x.shape == new_images[0].shape for x in new_images):
-        return torch.stack(new_images, dim=0)
-    return new_images
-
-def tokenizer_image_token(prompt: str, tokenizer: Any, image_token_index: int = IMAGE_TOKEN_INDEX, return_tensors: Optional[str] = None):
-    prompt_chunks = [tokenizer(chunk).input_ids for chunk in prompt.split("<image>")]
-
-    def insert_separator(X, sep):
-        return [ele for sublist in zip(X, [sep] * len(X)) for ele in sublist][:-1]
-
-    input_ids: List[int] = []
-    offset = 0
-    if len(prompt_chunks) > 0 and len(prompt_chunks[0]) > 0 and prompt_chunks[0][0] == tokenizer.bos_token_id:
-        offset = 1
-        input_ids.append(prompt_chunks[0][0])
-
-    for x in insert_separator(prompt_chunks, [image_token_index] * (offset + 1)):
-        input_ids.extend(x[offset:])
-
-    if return_tensors is not None:
-        if return_tensors == "pt":
-            return torch.tensor(input_ids, dtype=torch.long)
-        raise ValueError(f"Unsupported tensor type: {return_tensors}")
-    return input_ids
-
-def get_model_name_from_path(model_path: str) -> str:
-    model_path = model_path.strip("/")
-    model_paths = model_path.split("/")
-    if model_paths[-1].startswith("checkpoint-"):
-        return model_paths[-2] + "_" + model_paths[-1]
-    else:
-        return model_paths[-1]
-
-# Upstream ile uyumlu: durdurma kriteri
-class KeywordsStoppingCriteria(StoppingCriteria):
-    def __init__(self, keywords, tokenizer, input_ids):
-        self.keywords = keywords
-        self.keyword_ids = []
-        self.max_keyword_len = 0
-        for keyword in keywords:
-            cur_keyword_ids = tokenizer(keyword).input_ids
-            if len(cur_keyword_ids) > 1 and cur_keyword_ids[0] == tokenizer.bos_token_id:
-                cur_keyword_ids = cur_keyword_ids[1:]
-            if len(cur_keyword_ids) > self.max_keyword_len:
-                self.max_keyword_len = len(cur_keyword_ids)
-            self.keyword_ids.append(torch.tensor(cur_keyword_ids))
-        self.tokenizer = tokenizer
-        self.start_len = input_ids.shape[1]
-
-    def call_for_batch(self, output_ids: torch.LongTensor, scores: torch.FloatTensor, **kwargs) -> bool:
-        offset = min(output_ids.shape[1] - self.start_len, self.max_keyword_len)
-        self.keyword_ids = [kid.to(output_ids.device) for kid in self.keyword_ids]
-        for kid in self.keyword_ids:
-            truncated = output_ids[0, -kid.shape[0]:]
-            if torch.equal(truncated, kid):
-                return True
-        outputs = self.tokenizer.batch_decode(output_ids[:, -offset:], skip_special_tokens=True)[0]
-        for keyword in self.keywords:
-            if keyword in outputs:
-                return True
-        return False
-
-    def __call__(self, output_ids: torch.LongTensor, scores: torch.FloatTensor, **kwargs) -> bool:
-        outs = []
-        for i in range(output_ids.shape[0]):
-            outs.append(self.call_for_batch(output_ids[i].unsqueeze(0), scores))
-        return all(outs)