src/models/cnn_branch.py

"""
cnn_branch.py — ResNet-50 spatial feature extractor for satellite imagery.

Backbone: ResNet-50 pretrained on ImageNet (torchvision).
Modified to accept 4-channel input (RGB + NIR) by extending the first
convolutional layer. The final FC classification head is removed,
producing a 2048-dimensional spatial feature vector.
"""

import logging

import torch
import torch.nn as nn
from torchvision.models import resnet50, ResNet50_Weights

from src.training.config import IMAGE_CHANNELS, CNN_FEATURE_DIM

logger = logging.getLogger(__name__)


class CNNBranch(nn.Module):
    """
    ResNet-50 backbone modified for 4-channel satellite image input.

    Architecture:
        Input:  (batch, 4, 128, 128)
        Output: (batch, 2048) spatial feature vector

    The first conv layer is extended from 3→4 input channels by
    averaging the pretrained weights and appending them as the NIR
    channel weights. All subsequent layers use pretrained ImageNet weights.
    """

    def __init__(
        self,
        in_channels: int = IMAGE_CHANNELS,
        out_features: int = CNN_FEATURE_DIM,
        pretrained: bool = True,
        freeze_early: bool = False,
    ):
        """
        Args:
            in_channels: Number of input channels (4 for RGB+NIR).
            out_features: Output feature dimensionality (2048).
            pretrained: Whether to use ImageNet pretrained weights.
            freeze_early: Whether to freeze early layers (layer1, layer2).
        """
        super().__init__()

        # Load pretrained ResNet-50
        if pretrained:
            logger.info("Loading pretrained ResNet-50 weights (ImageNet)...")
            weights = ResNet50_Weights.IMAGENET1K_V2
            backbone = resnet50(weights=weights)
        else:
            backbone = resnet50(weights=None)

        # Modify first conv layer to accept 4 channels
        original_conv1 = backbone.conv1
        self.conv1 = nn.Conv2d(
            in_channels=in_channels,
            out_channels=64,
            kernel_size=7,
            stride=2,
            padding=3,
            bias=False,
        )

        # Initialize: copy pretrained RGB weights, average for NIR channel
        if pretrained and in_channels == 4:
            with torch.no_grad():
                # Copy RGB weights
                self.conv1.weight[:, :3, :, :] = original_conv1.weight
                # NIR channel: average of RGB weights
                self.conv1.weight[:, 3:, :, :] = original_conv1.weight.mean(dim=1, keepdim=True)
        elif pretrained and in_channels == 3:
            self.conv1.weight = original_conv1.weight

        # Copy remaining layers
        self.bn1 = backbone.bn1
        self.relu = backbone.relu
        self.maxpool = backbone.maxpool
        self.layer1 = backbone.layer1
        self.layer2 = backbone.layer2
        self.layer3 = backbone.layer3
        self.layer4 = backbone.layer4
        self.avgpool = backbone.avgpool

        # Optionally freeze early layers for transfer learning
        if freeze_early:
            for param in [self.conv1, self.bn1, self.layer1, self.layer2]:
                if isinstance(param, nn.Module):
                    for p in param.parameters():
                        p.requires_grad = False
            logger.info("Froze early layers (conv1, bn1, layer1, layer2).")

        self.out_features = out_features
        logger.info(f"CNNBranch initialized: {in_channels}ch → {out_features}d features")

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        Forward pass through ResNet-50 backbone.

        Args:
            x: Input tensor of shape (batch, 4, 128, 128).

        Returns:
            Feature vector of shape (batch, 2048).
        """
        x = self.conv1(x)
        x = self.bn1(x)
        x = self.relu(x)
        x = self.maxpool(x)

        x = self.layer1(x)
        x = self.layer2(x)
        x = self.layer3(x)
        x = self.layer4(x)  # (batch, 2048, H', W')

        x = self.avgpool(x)  # (batch, 2048, 1, 1)
        x = torch.flatten(x, 1)  # (batch, 2048)

        return x

    def get_feature_maps(self, x: torch.Tensor) -> torch.Tensor:
        """
        Get intermediate feature maps before global average pooling.
        Used by Grad-CAM for visualization.

        Returns:
            Feature maps of shape (batch, 2048, H', W').
        """
        x = self.conv1(x)
        x = self.bn1(x)
        x = self.relu(x)
        x = self.maxpool(x)

        x = self.layer1(x)
        x = self.layer2(x)
        x = self.layer3(x)
        x = self.layer4(x)

        return x  # (batch, 2048, 4, 4) for 128×128 input


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    model = CNNBranch(in_channels=4, pretrained=True)

    # Test forward pass
    dummy = torch.randn(2, 4, 128, 128)
    features = model(dummy)
    print(f"Input: {dummy.shape} → Output: {features.shape}")

    fmaps = model.get_feature_maps(dummy)
    print(f"Feature maps: {fmaps.shape}")

    # Parameter count
    total = sum(p.numel() for p in model.parameters())
    trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)
    print(f"Parameters: {total:,} total, {trainable:,} trainable")