scripts/sam3_buscot_runner.py

#!/usr/bin/env python3
from __future__ import annotations

import json
import math
import os
import sys
from pathlib import Path

import numpy as np
import torch
import torchvision.transforms.functional as TF
from PIL import Image

SCRIPT_DIR = Path(__file__).resolve().parent
BUNDLE_ROOT = SCRIPT_DIR.parent
SAM3_REPO = Path(os.environ.get("SAM3_REPO", BUNDLE_ROOT / "runtime" / "sam3_repo"))
for path in (SCRIPT_DIR, SAM3_REPO):
    if str(path) not in sys.path:
        sys.path.insert(0, str(path))

from sam3_decoder_experiment_lib import SAM3FeatureModel, resize_pad_image_and_mask  # noqa: E402


def bbox_xyxy_to_padded_cxcywh(box, orig_w: int, orig_h: int, image_size: int):
    x1, y1, x2, y2 = [float(v) for v in box]
    scale = float(image_size) / float(max(orig_h, orig_w))
    new_w = orig_w * scale
    new_h = orig_h * scale
    left = (image_size - new_w) / 2.0
    top = (image_size - new_h) / 2.0
    x1p, x2p = x1 * scale + left, x2 * scale + left
    y1p, y2p = y1 * scale + top, y2 * scale + top
    cx = (x1p + x2p) / 2.0 / image_size
    cy = (y1p + y2p) / 2.0 / image_size
    w = max(1.0, x2p - x1p) / image_size
    h = max(1.0, y2p - y1p) / image_size
    return torch.tensor([[[cx, cy, w, h]]], dtype=torch.float32)


def unpad_resize_prediction(prob_512: np.ndarray, orig_w: int, orig_h: int, image_size: int):
    scale = float(image_size) / float(max(orig_h, orig_w))
    new_w = max(1, int(round(orig_w * scale)))
    new_h = max(1, int(round(orig_h * scale)))
    left = (image_size - new_w) // 2
    top = (image_size - new_h) // 2
    crop = prob_512[top : top + new_h, left : left + new_w]
    return np.asarray(Image.fromarray(crop).resize((orig_w, orig_h), Image.BILINEAR))


class SAM3BuscotPredictor:
    def __init__(
        self,
        sam3_checkpoint: str,
        checkpoint_path: str | None = None,
        prompt_type: str = "semantic_text",
        prompt_text: str = "breast tumor",
        image_size: int = 512,
        device: str = "cuda",
        encoder_trainable: str = "frozen",
        lora_rank: int = 8,
        lora_alpha: float = 16,
        threshold: float = 0.5,
    ):
        self.image_size = int(image_size)
        self.threshold = float(threshold)
        self.device = torch.device(device if device == "cuda" and torch.cuda.is_available() else "cpu")
        self.model = SAM3FeatureModel(
            sam3_checkpoint,
            image_size=self.image_size,
            encoder_trainable=encoder_trainable,
            decoder_name="sam3_native",
            prompt_type=prompt_type,
            prompt_text=prompt_text,
            lora_rank=lora_rank,
            lora_alpha=lora_alpha,
        ).to(self.device)
        if checkpoint_path:
            ckpt = torch.load(checkpoint_path, map_location=self.device)
            state = ckpt["model"] if isinstance(ckpt, dict) and "model" in ckpt else ckpt
            cleaned = {}
            for k, v in state.items():
                for prefix in ("module.", "model."):
                    if k.startswith(prefix):
                        k = k[len(prefix) :]
                cleaned[k] = v
            self.model.load_state_dict(cleaned, strict=False)
        self.model.eval()

    @torch.no_grad()
    def predict(self, image_path: str, bbox_xyxy=None):
        image = Image.open(image_path).convert("RGB")
        orig_w, orig_h = image.size
        dummy = np.zeros((orig_h, orig_w), dtype=np.uint8)
        padded, _ = resize_pad_image_and_mask(image, dummy, self.image_size)
        x = TF.to_tensor(padded).unsqueeze(0).to(self.device)
        bbox_prompt = None
        if bbox_xyxy is not None:
            bbox_prompt = bbox_xyxy_to_padded_cxcywh(bbox_xyxy, orig_w, orig_h, self.image_size).to(self.device)
        with torch.amp.autocast(device_type="cuda", dtype=torch.bfloat16, enabled=self.device.type == "cuda"):
            raw = self.model(x, return_details=True, bbox_prompt=bbox_prompt)
        logits = raw["mask_logits"] if isinstance(raw, dict) else raw
        prob_512 = torch.sigmoid(logits)[0, 0].detach().float().cpu().numpy()
        prob = unpad_resize_prediction(prob_512, orig_w, orig_h, self.image_size)
        pred = prob >= self.threshold
        details = {}
        if isinstance(raw, dict):
            for key, value in raw.items():
                if key == "mask_logits":
                    continue
                try:
                    details[key] = float(torch.as_tensor(value).detach().flatten()[0].cpu())
                except Exception:
                    details[key] = str(value)
        return pred.astype(np.uint8), details