File size: 6,846 Bytes
b612017
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
"""PathAGG Hugging Face loaders (trust_remote_code).

Usage:
  from transformers import AutoModel
  model = AutoModel.from_pretrained("luoxd96/PathAGG", subfolder="vitb", trust_remote_code=True)
  cls = model(images)  # [B, D]
"""

from __future__ import annotations

from pathlib import Path
from typing import Any

import torch
import torch.nn as nn
from transformers import PretrainedConfig, PreTrainedModel
from transformers.utils import cached_file


class PathAGGConfig(PretrainedConfig):
    model_type = "pathagg"

    def __init__(
        self,
        variant: str = "vitb",
        img_size: int = 224,
        patch_size: int = 14,
        embed_dim: int = 768,
        depth: int = 12,
        num_heads: int = 12,
        num_register_tokens: int = 4,
        mlp_ratio: float = 4.0,
        qkv_bias: bool = True,
        init_values: float | None = None,
        no_embed_class: bool = False,
        return_patch: bool = False,
        **kwargs: Any,
    ) -> None:
        super().__init__(**kwargs)
        self.variant = variant
        self.img_size = img_size
        self.patch_size = patch_size
        self.embed_dim = embed_dim
        self.depth = depth
        self.num_heads = num_heads
        self.num_register_tokens = num_register_tokens
        self.mlp_ratio = mlp_ratio
        self.qkv_bias = qkv_bias
        self.init_values = init_values
        self.no_embed_class = no_embed_class
        self.return_patch = return_patch


def _build_backbone(config: PathAGGConfig) -> nn.Module:
    from timm.models.vision_transformer import VisionTransformer

    kwargs: dict[str, Any] = dict(
        img_size=config.img_size,
        patch_size=config.patch_size,
        in_chans=3,
        num_classes=0,
        global_pool="",
        embed_dim=config.embed_dim,
        depth=config.depth,
        num_heads=config.num_heads,
        mlp_ratio=config.mlp_ratio,
        qkv_bias=config.qkv_bias,
        reg_tokens=config.num_register_tokens,
    )
    if config.init_values is not None:
        kwargs["init_values"] = config.init_values
    if config.no_embed_class:
        kwargs["no_embed_class"] = True
    return VisionTransformer(**kwargs)


class PathAGGModel(PreTrainedModel):
    """Pathology multi-teacher KD student. Forward returns CLS ``[B, D]`` by default."""

    config_class = PathAGGConfig
    base_model_prefix = "pathagg"
    _no_split_modules = ["Block"]

    def __init__(self, config: PathAGGConfig) -> None:
        super().__init__(config)
        self.return_patch = bool(config.return_patch)
        self.model = _build_backbone(config)
        # Do not call post_init() random re-init; weights come from checkpoint.

    @property
    def embed_dim(self) -> int:
        return int(self.model.embed_dim)

    @property
    def num_register_tokens(self) -> int:
        return int(getattr(self.model, "reg_tokens", getattr(self.model, "num_register_tokens", 4)))

    def forward_features(self, x: torch.Tensor) -> torch.Tensor:
        return self.model.forward_features(x)

    def forward(
        self,
        pixel_values: torch.Tensor | None = None,
        x: torch.Tensor | None = None,
        return_patch: bool | None = None,
        **kwargs: Any,
    ):
        if pixel_values is None and x is None:
            raise ValueError("provide pixel_values=... or x=...")
        images = pixel_values if pixel_values is not None else x
        tokens = self.forward_features(images)
        cls = tokens[:, 0]
        use_patch = self.return_patch if return_patch is None else bool(return_patch)
        if not use_patch:
            return cls
        num_prefix = int(getattr(self.model, "num_prefix_tokens", 1 + self.num_register_tokens))
        patch = tokens[:, num_prefix:]
        return cls, patch

    @classmethod
    def from_pretrained(cls, pretrained_model_name_or_path, *model_args, **kwargs):
        """Load config + weights without transformers gamma↔weight remapping (LayerScale)."""
        subfolder = kwargs.pop("subfolder", "")
        local_files_only = kwargs.pop("local_files_only", False)
        revision = kwargs.pop("revision", None)
        cache_dir = kwargs.pop("cache_dir", None)
        token = kwargs.pop("token", None)
        if token is None:
            token = kwargs.pop("use_auth_token", None)
        kwargs.pop("trust_remote_code", None)
        kwargs.pop("torch_dtype", None)
        kwargs.pop("device_map", None)
        kwargs.pop("low_cpu_mem_usage", None)
        return_patch = kwargs.pop("return_patch", None)
        config = kwargs.pop("config", None)

        if config is None:
            config = PathAGGConfig.from_pretrained(
                pretrained_model_name_or_path,
                subfolder=subfolder,
                local_files_only=local_files_only,
                revision=revision,
                cache_dir=cache_dir,
                token=token,
                **kwargs,
            )
        if return_patch is not None:
            config.return_patch = bool(return_patch)

        model = cls(config)

        weight_file = cached_file(
            pretrained_model_name_or_path,
            "pytorch_model.bin",
            subfolder=subfolder,
            local_files_only=local_files_only,
            revision=revision,
            cache_dir=cache_dir,
            token=token,
        )
        if weight_file is None:
            raise FileNotFoundError(
                f"pytorch_model.bin not found under {pretrained_model_name_or_path!r} (subfolder={subfolder!r})"
            )
        try:
            state = torch.load(weight_file, map_location="cpu", weights_only=True)
        except TypeError:
            state = torch.load(weight_file, map_location="cpu")

        # transformers may have renamed LayerScale gamma→weight in some pipelines; normalize back.
        fixed = {}
        for k, v in state.items():
            if k.endswith(".ls1.weight") or k.endswith(".ls2.weight"):
                fixed[k[: -len(".weight")] + ".gamma"] = v
            else:
                fixed[k] = v
        missing, unexpected = model.load_state_dict(fixed, strict=True)
        if missing or unexpected:
            raise RuntimeError(f"load failed: missing={missing}, unexpected={unexpected}")
        model.eval()
        return model


def get_preprocess(img_size: int = 224):
    """ImageNet normalize preprocess (PIL → tensor)."""
    from torchvision import transforms

    return transforms.Compose(
        [
            transforms.Resize(img_size, interpolation=transforms.InterpolationMode.BICUBIC),
            transforms.CenterCrop(img_size),
            transforms.ToTensor(),
            transforms.Normalize(mean=(0.485, 0.456, 0.406), std=(0.229, 0.224, 0.225)),
        ]
    )


__all__ = [
    "PathAGGConfig",
    "PathAGGModel",
    "get_preprocess",
]