Added code
This commit is contained in:
175
code/src/model.py
Normal file
175
code/src/model.py
Normal file
@@ -0,0 +1,175 @@
|
||||
from __future__ import annotations
|
||||
|
||||
"""EdgeNeXt builder + transfer-learning regimes (freeze / partial / MONA).
|
||||
|
||||
Three fine-tuning regimes are exposed (see PROTOCOL §5):
|
||||
* ``full`` — train the whole backbone + head (one or two LR groups).
|
||||
* ``partial`` — freeze the first ``4 - n_unfrozen`` stages, train the rest + head;
|
||||
``n_unfrozen=0`` is pure linear probing (feature extraction).
|
||||
* ``mona`` — freeze the whole backbone, train only MONA adapters + head (PEFT).
|
||||
|
||||
``feature_extract_features`` returns L2-normalized pooled embeddings for the
|
||||
UMAP analysis (see ``umap_analysis.py``).
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
import timm
|
||||
import torch
|
||||
import torch.nn.functional as F
|
||||
from torch import nn
|
||||
|
||||
from .mona import attach_mona_adapters
|
||||
|
||||
|
||||
@dataclass
|
||||
class ModelConfig:
|
||||
"""Backbone + regime configuration (a subset is searched by Optuna)."""
|
||||
|
||||
model_name: str = "edgenext_small.usi_in1k"
|
||||
num_classes: int = 13 # Naruto Sign: 12 seals + 'zero' (verify after download)
|
||||
pretrained: bool = True
|
||||
drop_rate: float = 0.0 # classifier dropout
|
||||
drop_path_rate: float = 0.0 # stochastic depth
|
||||
regime: str = "partial" # {"full", "partial", "mona"}
|
||||
n_unfrozen_stages: int = 1 # used by "partial": 0..4
|
||||
mona_stages: tuple[int, ...] = (2, 3) # used by "mona"
|
||||
mona_bottleneck: int = 64
|
||||
mona_kernels: tuple[int, ...] = (3, 5, 7)
|
||||
|
||||
|
||||
def _set_norm_eval(module: nn.Module) -> None:
|
||||
"""Put frozen BatchNorm/LayerNorm/GroupNorm in eval mode (freeze stats)."""
|
||||
for m in module.modules():
|
||||
if isinstance(m, (nn.BatchNorm2d, nn.LayerNorm, nn.GroupNorm)):
|
||||
m.eval()
|
||||
|
||||
|
||||
def build_model(cfg: ModelConfig) -> nn.Module:
|
||||
"""Create an EdgeNeXt and apply the requested fine-tuning regime.
|
||||
|
||||
Args:
|
||||
cfg: Model configuration.
|
||||
|
||||
Returns:
|
||||
The configured ``nn.Module``. Trainable parameters depend on ``regime``.
|
||||
"""
|
||||
model = timm.create_model(
|
||||
cfg.model_name,
|
||||
pretrained=cfg.pretrained,
|
||||
num_classes=cfg.num_classes,
|
||||
drop_rate=cfg.drop_rate,
|
||||
drop_path_rate=cfg.drop_path_rate,
|
||||
)
|
||||
|
||||
if cfg.regime == "full":
|
||||
return model
|
||||
|
||||
# Freeze the whole backbone first; head stays trainable in every regime.
|
||||
for name, p in model.named_parameters():
|
||||
if not _is_head(name):
|
||||
p.requires_grad_(False)
|
||||
|
||||
if cfg.regime == "partial":
|
||||
n = max(0, min(cfg.n_unfrozen_stages, len(model.stages)))
|
||||
if n > 0:
|
||||
for idx in range(len(model.stages) - n, len(model.stages)):
|
||||
for p in model.stages[idx].parameters():
|
||||
p.requires_grad_(True)
|
||||
# final norm before the head (if present) follows the head.
|
||||
for name, p in model.named_parameters():
|
||||
if "norm_pre" in name or name.startswith("norm"):
|
||||
p.requires_grad_(True)
|
||||
elif cfg.regime == "mona":
|
||||
attach_mona_adapters(
|
||||
model,
|
||||
list(cfg.mona_stages),
|
||||
bottleneck=cfg.mona_bottleneck,
|
||||
kernels=cfg.mona_kernels,
|
||||
) # adapters are created trainable by default
|
||||
else:
|
||||
raise ValueError(f"unknown regime: {cfg.regime!r}")
|
||||
|
||||
return model
|
||||
|
||||
|
||||
def _is_head(param_name: str) -> bool:
|
||||
"""Heuristic: classifier-head parameters in timm models contain 'head'/'fc'."""
|
||||
return "head" in param_name or param_name.startswith("fc")
|
||||
|
||||
|
||||
def freeze_consistency(model: nn.Module) -> None:
|
||||
"""Set frozen norm layers to eval so their running stats are not updated.
|
||||
|
||||
Call this **after** ``model.train()`` in every training step for the
|
||||
``partial``/``mona`` regimes (see PROTOCOL §5.2 — the classic BN pitfall).
|
||||
"""
|
||||
for module in model.modules():
|
||||
if isinstance(module, (nn.BatchNorm2d, nn.LayerNorm, nn.GroupNorm)):
|
||||
if not any(p.requires_grad for p in module.parameters(recurse=False)):
|
||||
module.eval()
|
||||
|
||||
|
||||
def build_param_groups(
|
||||
model: nn.Module, base_lr: float, *, backbone_lr_mult: float = 0.1
|
||||
) -> list[dict]:
|
||||
"""Two LR groups: a smaller LR for backbone, the base LR for head/adapters.
|
||||
|
||||
Discriminative learning rates (lower layers -> lower LR) are a standard
|
||||
transfer-learning trick (see PROTOCOL §5.3). Only parameters with
|
||||
``requires_grad=True`` are included.
|
||||
|
||||
Args:
|
||||
model: The configured model.
|
||||
base_lr: LR for the head and MONA adapters.
|
||||
backbone_lr_mult: Multiplier applied to ``base_lr`` for backbone params.
|
||||
|
||||
Returns:
|
||||
A list of optimizer parameter-group dicts.
|
||||
"""
|
||||
head, backbone = [], []
|
||||
for name, p in model.named_parameters():
|
||||
if not p.requires_grad:
|
||||
continue
|
||||
if _is_head(name) or "mona_adapters" in name:
|
||||
head.append(p)
|
||||
else:
|
||||
backbone.append(p)
|
||||
|
||||
groups: list[dict] = []
|
||||
if head:
|
||||
groups.append({"params": head, "lr": base_lr})
|
||||
if backbone:
|
||||
groups.append({"params": backbone, "lr": base_lr * backbone_lr_mult})
|
||||
return groups
|
||||
|
||||
|
||||
def count_trainable(model: nn.Module) -> tuple[int, int]:
|
||||
"""Return ``(trainable, total)`` parameter counts."""
|
||||
total = sum(p.numel() for p in model.parameters())
|
||||
trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)
|
||||
return trainable, total
|
||||
|
||||
|
||||
@torch.inference_mode()
|
||||
def extract_features(
|
||||
model: nn.Module, x: torch.Tensor, *, l2_normalize: bool = True
|
||||
) -> torch.Tensor:
|
||||
"""Return pooled pre-logit embeddings ``[B, num_features]`` for UMAP.
|
||||
|
||||
Uses timm's ``forward_features`` + ``forward_head(..., pre_logits=True)`` so
|
||||
the classifier is bypassed.
|
||||
|
||||
Args:
|
||||
model: A timm model in eval mode.
|
||||
x: Input batch ``[B, 3, H, W]`` already normalized.
|
||||
l2_normalize: If ``True`` L2-normalize embeddings (cosine geometry).
|
||||
|
||||
Returns:
|
||||
Float tensor on CPU of shape ``[B, num_features]``.
|
||||
"""
|
||||
feats = model.forward_features(x)
|
||||
emb = model.forward_head(feats, pre_logits=True)
|
||||
if l2_normalize:
|
||||
emb = F.normalize(emb, dim=1)
|
||||
return emb.detach().cpu()
|
||||
Reference in New Issue
Block a user