utils.py

import math
import warnings

import numpy as np
from PIL import Image
import timm
import torch

from models import mocov3_vit


def fix_mocov3_state_dict(state_dict):
    for k in list(state_dict.keys()):
        # retain only base_encoder up to before the embedding layer
        if k.startswith('module.base_encoder'):
            # fix naming bug in checkpoint
            new_k = k[len("module.base_encoder."):]
            if "blocks.13.norm13" in new_k:
                new_k = new_k.replace("norm13", "norm1")
            if "blocks.13.mlp.fc13" in k:
                new_k = new_k.replace("fc13", "fc1")
            if "blocks.14.norm14" in k:
                new_k = new_k.replace("norm14", "norm2")
            if "blocks.14.mlp.fc14" in k:
                new_k = new_k.replace("fc14", "fc2")
            # remove prefix
            if 'head' not in new_k and new_k.split('.')[0] != 'fc':
                state_dict[new_k] = state_dict[k]
        # delete renamed or unused k
        del state_dict[k]
    if 'pos_embed' in state_dict.keys():
        state_dict['pos_embed'] = timm.layers.pos_embed.resample_abs_pos_embed(
            state_dict['pos_embed'], [16, 16],
        )
    return state_dict

@torch.no_grad()
def load_encoders(enc_type, device, resolution=256):
    assert (resolution == 256) or (resolution == 512)
    
    enc_names = enc_type.split(',')
    encoders, architectures, encoder_types = [], [], []
    for enc_name in enc_names:
        encoder_type, architecture, model_config = enc_name.split('-')
        # Currently, we only support 512x512 experiments with DINOv2 encoders.
        if resolution == 512:
            if encoder_type != 'dinov2':
                raise NotImplementedError(
                    "Currently, we only support 512x512 experiments with DINOv2 encoders."
                    )

        architectures.append(architecture)
        encoder_types.append(encoder_type)
        if encoder_type == 'mocov3':
            if architecture == 'vit':
                if model_config == 's':
                    encoder = mocov3_vit.vit_small()
                elif model_config == 'b':
                    encoder = mocov3_vit.vit_base()
                elif model_config == 'l':
                    encoder = mocov3_vit.vit_large()
                ckpt = torch.load(f'./ckpts/mocov3_vit{model_config}.pth')
                state_dict = fix_mocov3_state_dict(ckpt['state_dict'])
                del encoder.head
                encoder.load_state_dict(state_dict, strict=True)
                encoder.head = torch.nn.Identity()
            elif architecture == 'resnet':
                raise NotImplementedError()
 
            encoder = encoder.to(device)
            encoder.eval()

        elif 'dinov2' in encoder_type:
            import timm
            if 'reg' in encoder_type:
                encoder = torch.hub.load('facebookresearch/dinov2', f'dinov2_vit{model_config}14_reg')
            else:
                encoder = torch.hub.load('facebookresearch/dinov2', f'dinov2_vit{model_config}14')
            del encoder.head
            patch_resolution = 16 * (resolution // 256)
            encoder.pos_embed.data = timm.layers.pos_embed.resample_abs_pos_embed(
                encoder.pos_embed.data, [patch_resolution, patch_resolution],
            )
            encoder.head = torch.nn.Identity()
            encoder = encoder.to(device)
            encoder.eval()
        
        elif 'dinov1' == encoder_type:
            import timm
            from models import dinov1
            encoder = dinov1.vit_base()
            ckpt =  torch.load(f'./ckpts/dinov1_vit{model_config}.pth') 
            if 'pos_embed' in ckpt.keys():
                ckpt['pos_embed'] = timm.layers.pos_embed.resample_abs_pos_embed(
                    ckpt['pos_embed'], [16, 16],
                )
            del encoder.head
            encoder.head = torch.nn.Identity()
            encoder.load_state_dict(ckpt, strict=True)
            encoder = encoder.to(device)
            encoder.forward_features = encoder.forward
            encoder.eval()

        elif encoder_type == 'clip':
            import clip
            from models.clip_vit import UpdatedVisionTransformer
            encoder_ = clip.load(f"ViT-{model_config}/14", device='cpu')[0].visual
            encoder = UpdatedVisionTransformer(encoder_).to(device)
             #.to(device)
            encoder.embed_dim = encoder.model.transformer.width
            encoder.forward_features = encoder.forward
            encoder.eval()
        
        elif encoder_type == 'mae':
            from models.mae_vit import vit_large_patch16
            import timm
            kwargs = dict(img_size=256)
            encoder = vit_large_patch16(**kwargs).to(device)
            with open(f"ckpts/mae_vit{model_config}.pth", "rb") as f:
                state_dict = torch.load(f)
            if 'pos_embed' in state_dict["model"].keys():
                state_dict["model"]['pos_embed'] = timm.layers.pos_embed.resample_abs_pos_embed(
                    state_dict["model"]['pos_embed'], [16, 16],
                )
            encoder.load_state_dict(state_dict["model"])

            encoder.pos_embed.data = timm.layers.pos_embed.resample_abs_pos_embed(
                encoder.pos_embed.data, [16, 16],
            )

        elif encoder_type == 'jepa':
            from models.jepa import vit_huge
            kwargs = dict(img_size=[224, 224], patch_size=14)
            encoder = vit_huge(**kwargs).to(device)
            with open(f"ckpts/ijepa_vit{model_config}.pth", "rb") as f:
                state_dict = torch.load(f, map_location=device)
            new_state_dict = dict()
            for key, value in state_dict['encoder'].items():
                new_state_dict[key[7:]] = value
            encoder.load_state_dict(new_state_dict)
            encoder.forward_features = encoder.forward

        # ------------------------------------------------------------------
        # Pathology Foundation Models
        # ------------------------------------------------------------------
        elif encoder_type == 'uni':
            # UNI (MahmoodLab/UNI) — ViT-L/16 trained on pathology images
            # Checkpoint: place at ckpts/uni.bin (from HuggingFace hub)
            import timm
            encoder = timm.create_model(
                "vit_large_patch16_224",
                img_size=224,
                patch_size=16,
                init_values=1e-5,
                num_classes=0,
                dynamic_img_size=True,
            )
            state_dict = torch.load("ckpts/uni.bin", map_location="cpu")
            encoder.load_state_dict(state_dict, strict=True)
            # embed_dim = 1024 (ViT-L)
            encoder.forward_features = encoder.forward_features
            encoder = encoder.to(device)
            encoder.eval()

        elif encoder_type == 'uni2':
            # UNI2-h (MahmoodLab/UNI2-h) — ViT-H/14 trained on 100M+ pathology images
            # Checkpoint: place at ckpts/pytorch_model.bin (from HuggingFace hub)
            import timm
            import os
            
            timm_kwargs = {
                'img_size': 224,
                'patch_size': 14,
                'depth': 24,
                'num_heads': 24,
                'init_values': 1e-5,
                'embed_dim': 1536,
                'mlp_ratio': 2.66667 * 2,
                'num_classes': 0,
                'no_embed_class': True,
                'mlp_layer': timm.layers.SwiGLUPacked,
                'act_layer': 'silu',
                'reg_tokens': 8,
                'dynamic_img_size': True
            }
            
            local_ckpt = "ckpts/pytorch_model.bin"
            if os.path.exists(local_ckpt):
                encoder = timm.create_model("hf_hub:MahmoodLab/UNI2-h", pretrained=False, **timm_kwargs)
                state_dict = torch.load(local_ckpt, map_location="cpu")
                encoder.load_state_dict(state_dict, strict=True)
            else:
                try:
                    encoder = timm.create_model("hf_hub:MahmoodLab/UNI2-h", pretrained=True, **timm_kwargs)
                except Exception as e:
                    raise FileNotFoundError(
                        f"Failed to load UNI2-h from Hugging Face Hub, and local checkpoint '{local_ckpt}' was not found. "
                        f"If you downloaded it manually, please place it at '{local_ckpt}'. Error details: {str(e)}"
                    )
            
            encoder.embed_dim = 1536
            _uni2_ff = encoder.forward_features
            def _uni2_forward_features(x, _ff=_uni2_ff):
                feats = _ff(x)
                if feats.shape[1] == 265:
                    return feats[:, 1:257] # drop CLS token at index 0 and 8 register tokens at the end
                elif feats.shape[1] == 264:
                    return feats[:, :256] # drop 8 register tokens
                return feats
            
            encoder.forward_features = _uni2_forward_features
            encoder = encoder.to(device)
            encoder.eval()

        elif encoder_type == 'conch':
            # CONCH (MahmoodLab/CONCH) — vision-language model for pathology
            # Install: pip install conch
            # Checkpoint: place at ckpts/conch.pt
            try:
                from conch.open_clip_custom import create_model_from_pretrained
                _conch_raw, _ = create_model_from_pretrained(
                    "conch_ViT-B-16", "ckpts/conch.pt"
                )
                class _CONCHWrapper(torch.nn.Module):
                    def __init__(self, m):
                        super().__init__()
                        self.model = m
                        self.embed_dim = m.visual.output_dim
                    def forward_features(self, x):
                        return self.model.encode_image(
                            x, proj_contrast=False, normalize=False
                        )
                encoder = _CONCHWrapper(_conch_raw).to(device)
            except ImportError:
                raise ImportError(
                    "CONCH requires the `conch` package. "
                    "See: https://github.com/mahmoodlab/CONCH"
                )
            encoder.eval()

        elif encoder_type == 'gigapath':
            # GigaPath (prov-gigapath) — ViT-G/14 trained on 1.3B pathology patches
            # Install: pip install timm>=1.0.3
            # Checkpoint: place at ckpts/gigapath.pt
            import timm
            encoder = timm.create_model(
                "hf_hub:prov-gigapath/prov-gigapath",
                pretrained=False,
            )
            state_dict = torch.load("ckpts/gigapath.pt", map_location="cpu")
            encoder.load_state_dict(state_dict, strict=True)
            encoder.embed_dim = 1536  # ViT-G hidden dim
            _gp_ff = encoder.forward_features
            def _gigapath_forward_features(x, _ff=_gp_ff):
                feats = _ff(x)           # [B, N+1, D]
                return feats[:, 1:]      # drop CLS token, return patch tokens
            encoder.forward_features = _gigapath_forward_features
            encoder = encoder.to(device)
            encoder.eval()

        encoders.append(encoder)

    return encoders, encoder_types, architectures


def _no_grad_trunc_normal_(tensor, mean, std, a, b):
    # Cut & paste from PyTorch official master until it's in a few official releases - RW
    # Method based on https://people.sc.fsu.edu/~jburkardt/presentations/truncated_normal.pdf
    def norm_cdf(x):
        # Computes standard normal cumulative distribution function
        return (1. + math.erf(x / math.sqrt(2.))) / 2.

    if (mean < a - 2 * std) or (mean > b + 2 * std):
        warnings.warn("mean is more than 2 std from [a, b] in nn.init.trunc_normal_. "
                      "The distribution of values may be incorrect.",
                      stacklevel=2)

    with torch.no_grad():
        # Values are generated by using a truncated uniform distribution and
        # then using the inverse CDF for the normal distribution.
        # Get upper and lower cdf values
        l = norm_cdf((a - mean) / std)
        u = norm_cdf((b - mean) / std)

        # Uniformly fill tensor with values from [l, u], then translate to
        # [2l-1, 2u-1].
        tensor.uniform_(2 * l - 1, 2 * u - 1)

        # Use inverse cdf transform for normal distribution to get truncated
        # standard normal
        tensor.erfinv_()

        # Transform to proper mean, std
        tensor.mul_(std * math.sqrt(2.))
        tensor.add_(mean)

        # Clamp to ensure it's in the proper range
        tensor.clamp_(min=a, max=b)
        return tensor


def trunc_normal_(tensor, mean=0., std=1., a=-2., b=2.):
    return _no_grad_trunc_normal_(tensor, mean, std, a, b)


def center_crop_arr(image_arr, image_size):
    """
    Center cropping implementation from ADM.
    https://github.com/openai/guided-diffusion/blob/8fb3ad9197f16bbc40620447b2742e13458d2831/guided_diffusion/image_datasets.py#L126
    """
    pil_image = Image.fromarray(image_arr)
    while min(*pil_image.size) >= 2 * image_size:
        pil_image = pil_image.resize(
            tuple(x // 2 for x in pil_image.size), resample=Image.BOX
        )

    scale = image_size / min(*pil_image.size)
    pil_image = pil_image.resize(
        tuple(round(x * scale) for x in pil_image.size), resample=Image.BICUBIC
    )

    arr = np.array(pil_image)
    crop_y = (arr.shape[0] - image_size) // 2
    crop_x = (arr.shape[1] - image_size) // 2
    return arr[crop_y: crop_y + image_size, crop_x: crop_x + image_size]


def preprocess_imgs_vae(imgs):
    # imgs: (B, C, H, W) -> (B, C, H, W), [0, 255] uint8 -> [-1, 1] float32
    return imgs.float() / 127.5 - 1.


def count_trainable_params(m):
    return sum(p.numel() for p in m.parameters() if p.requires_grad)


def normalize_latents(latents, latents_scale, latents_bias):
    return (latents - latents_bias) * latents_scale


def denormalize_latents(latents, latents_scale, latents_bias):
    return latents / latents_scale + latents_bias