cleanup: remove pann/timm/bert/bart dead branches from model.py

models/CLAP/open_clip/model.py

@@ -14,13 +14,11 @@ import torch
 import torch.nn.functional as F
 from torch import nn
 
-from .timm_model import TimmModel
 import logging
 from .utils import freeze_batch_norm_2d
-
-from .pann_model import create_pann_model
 from .htsat import create_htsat_model
-from transformers import BertModel, RobertaModel, BartModel, RobertaConfig
+from transformers import RobertaModel, RobertaConfig
+# removed: TimmModel, create_pann_model, BertModel, BartModel — only HTSAT audio + roberta text used in inference
 from transformers.tokenization_utils_base import BatchEncoding
 
 
@@ -458,17 +456,8 @@ class CLAP(nn.Module):
         else:
             raise NotImplementedError
 
-        # audio branch
-        # audio branch parameters
-        if audio_cfg.model_type == "PANN":
-            self.audio_branch = create_pann_model(audio_cfg, enable_fusion, fusion_type)
-        elif audio_cfg.model_type == "HTSAT":
-            self.audio_branch = create_htsat_model(
-                audio_cfg, enable_fusion, fusion_type
-            )
-        else:
-            logging.error(f"Model config for {audio_cfg.model_type} not found")
-            raise RuntimeError(f"Model config for {audio_cfg.model_type} not found.")
+        # audio branch — removed: PANN branch, only HTSAT used
+        self.audio_branch = create_htsat_model(audio_cfg, enable_fusion, fusion_type)
 
         # text branch
         # text branch parameters
@@ -498,21 +487,7 @@ class CLAP(nn.Module):
                 mlp_act_layer,
                 nn.Linear(self.joint_embed_shape, self.joint_embed_shape),
             )
-        elif text_cfg.model_type == "bert":
-            self.text_branch = BertModel.from_pretrained("bert-base-uncased")
-            self.text_transform = MLPLayers(
-                units=[
-                    self.joint_embed_shape,
-                    self.joint_embed_shape,
-                    self.joint_embed_shape,
-                ],
-                dropout=0.1,
-            )
-            self.text_projection = nn.Sequential(
-                nn.Linear(768, self.joint_embed_shape),
-                mlp_act_layer,
-                nn.Linear(self.joint_embed_shape, self.joint_embed_shape),
-            )
+        # removed: bert branch — only roberta used
         elif text_cfg.model_type == "roberta":
             self.text_branch = RobertaModel.from_pretrained("roberta-base")
 
@@ -529,21 +504,7 @@ class CLAP(nn.Module):
                 mlp_act_layer,
                 nn.Linear(self.joint_embed_shape, self.joint_embed_shape),
             )
-        elif text_cfg.model_type == "bart":
-            self.text_branch = BartModel.from_pretrained("facebook/bart-base")
-            self.text_transform = MLPLayers(
-                units=[
-                    self.joint_embed_shape,
-                    self.joint_embed_shape,
-                    self.joint_embed_shape,
-                ],
-                dropout=0.1,
-            )
-            self.text_projection = nn.Sequential(
-                nn.Linear(768, self.joint_embed_shape),
-                mlp_act_layer,
-                nn.Linear(self.joint_embed_shape, self.joint_embed_shape),
-            )
+        # removed: bart branch — only roberta used
         else:
             logging.error(f"Model config for {text_cfg.model_type} not found")
             raise RuntimeError(f"Model config for {text_cfg.model_type} not found.")

models/CLAP/open_clip/pann_model.py

@@ -1,704 +0,0 @@
-# PANNs: Large-Scale Pretrained Audio Neural Networks for Audio Pattern Recognition
-# Reference from https://github.com/qiuqiangkong/audioset_tagging_cnn
-# Some layers are re-designed for CLAP
-import os
-
-os.environ["NUMBA_CACHE_DIR"] = "/tmp/"
-
-import torch
-import torch.nn as nn
-import torch.nn.functional as F
-from torchlibrosa.stft import Spectrogram, LogmelFilterBank
-from torchlibrosa.augmentation import SpecAugmentation
-
-from .utils import do_mixup, interpolate, pad_framewise_output
-from .feature_fusion import iAFF, AFF, DAF
-
-
-def init_layer(layer):
-    """Initialize a Linear or Convolutional layer."""
-    nn.init.xavier_uniform_(layer.weight)
-
-    if hasattr(layer, "bias"):
-        if layer.bias is not None:
-            layer.bias.data.fill_(0.0)
-
-
-def init_bn(bn):
-    """Initialize a Batchnorm layer."""
-    bn.bias.data.fill_(0.0)
-    bn.weight.data.fill_(1.0)
-
-
-class ConvBlock(nn.Module):
-    def __init__(self, in_channels, out_channels):
-
-        super(ConvBlock, self).__init__()
-
-        self.conv1 = nn.Conv2d(
-            in_channels=in_channels,
-            out_channels=out_channels,
-            kernel_size=(3, 3),
-            stride=(1, 1),
-            padding=(1, 1),
-            bias=False,
-        )
-
-        self.conv2 = nn.Conv2d(
-            in_channels=out_channels,
-            out_channels=out_channels,
-            kernel_size=(3, 3),
-            stride=(1, 1),
-            padding=(1, 1),
-            bias=False,
-        )
-
-        self.bn1 = nn.BatchNorm2d(out_channels)
-        self.bn2 = nn.BatchNorm2d(out_channels)
-
-        self.init_weight()
-
-    def init_weight(self):
-        init_layer(self.conv1)
-        init_layer(self.conv2)
-        init_bn(self.bn1)
-        init_bn(self.bn2)
-
-    def forward(self, input, pool_size=(2, 2), pool_type="avg"):
-
-        x = input
-        x = F.relu_(self.bn1(self.conv1(x)))
-        x = F.relu_(self.bn2(self.conv2(x)))
-        if pool_type == "max":
-            x = F.max_pool2d(x, kernel_size=pool_size)
-        elif pool_type == "avg":
-            x = F.avg_pool2d(x, kernel_size=pool_size)
-        elif pool_type == "avg+max":
-            x1 = F.avg_pool2d(x, kernel_size=pool_size)
-            x2 = F.max_pool2d(x, kernel_size=pool_size)
-            x = x1 + x2
-        else:
-            raise Exception("Incorrect argument!")
-
-        return x
-
-
-class ConvBlock5x5(nn.Module):
-    def __init__(self, in_channels, out_channels):
-
-        super(ConvBlock5x5, self).__init__()
-
-        self.conv1 = nn.Conv2d(
-            in_channels=in_channels,
-            out_channels=out_channels,
-            kernel_size=(5, 5),
-            stride=(1, 1),
-            padding=(2, 2),
-            bias=False,
-        )
-
-        self.bn1 = nn.BatchNorm2d(out_channels)
-
-        self.init_weight()
-
-    def init_weight(self):
-        init_layer(self.conv1)
-        init_bn(self.bn1)
-
-    def forward(self, input, pool_size=(2, 2), pool_type="avg"):
-
-        x = input
-        x = F.relu_(self.bn1(self.conv1(x)))
-        if pool_type == "max":
-            x = F.max_pool2d(x, kernel_size=pool_size)
-        elif pool_type == "avg":
-            x = F.avg_pool2d(x, kernel_size=pool_size)
-        elif pool_type == "avg+max":
-            x1 = F.avg_pool2d(x, kernel_size=pool_size)
-            x2 = F.max_pool2d(x, kernel_size=pool_size)
-            x = x1 + x2
-        else:
-            raise Exception("Incorrect argument!")
-
-        return x
-
-
-class AttBlock(nn.Module):
-    def __init__(self, n_in, n_out, activation="linear", temperature=1.0):
-        super(AttBlock, self).__init__()
-
-        self.activation = activation
-        self.temperature = temperature
-        self.att = nn.Conv1d(
-            in_channels=n_in,
-            out_channels=n_out,
-            kernel_size=1,
-            stride=1,
-            padding=0,
-            bias=True,
-        )
-        self.cla = nn.Conv1d(
-            in_channels=n_in,
-            out_channels=n_out,
-            kernel_size=1,
-            stride=1,
-            padding=0,
-            bias=True,
-        )
-
-        self.bn_att = nn.BatchNorm1d(n_out)
-        self.init_weights()
-
-    def init_weights(self):
-        init_layer(self.att)
-        init_layer(self.cla)
-        init_bn(self.bn_att)
-
-    def forward(self, x):
-        # x: (n_samples, n_in, n_time)
-        norm_att = torch.softmax(torch.clamp(self.att(x), -10, 10), dim=-1)
-        cla = self.nonlinear_transform(self.cla(x))
-        x = torch.sum(norm_att * cla, dim=2)
-        return x, norm_att, cla
-
-    def nonlinear_transform(self, x):
-        if self.activation == "linear":
-            return x
-        elif self.activation == "sigmoid":
-            return torch.sigmoid(x)
-
-
-class Cnn14(nn.Module):
-    def __init__(
-        self,
-        sample_rate,
-        window_size,
-        hop_size,
-        mel_bins,
-        fmin,
-        fmax,
-        classes_num,
-        enable_fusion=False,
-        fusion_type="None",
-    ):
-
-        super(Cnn14, self).__init__()
-
-        window = "hann"
-        center = True
-        pad_mode = "reflect"
-        ref = 1.0
-        amin = 1e-10
-        top_db = None
-
-        self.enable_fusion = enable_fusion
-        self.fusion_type = fusion_type
-
-        # Spectrogram extractor
-        self.spectrogram_extractor = Spectrogram(
-            n_fft=window_size,
-            hop_length=hop_size,
-            win_length=window_size,
-            window=window,
-            center=center,
-            pad_mode=pad_mode,
-            freeze_parameters=True,
-        )
-
-        # Logmel feature extractor
-        self.logmel_extractor = LogmelFilterBank(
-            sr=sample_rate,
-            n_fft=window_size,
-            n_mels=mel_bins,
-            fmin=fmin,
-            fmax=fmax,
-            ref=ref,
-            amin=amin,
-            top_db=top_db,
-            freeze_parameters=True,
-        )
-
-        # Spec augmenter
-        self.spec_augmenter = SpecAugmentation(
-            time_drop_width=64,
-            time_stripes_num=2,
-            freq_drop_width=8,
-            freq_stripes_num=2,
-        )
-
-        self.bn0 = nn.BatchNorm2d(64)
-
-        if (self.enable_fusion) and (self.fusion_type == "channel_map"):
-            self.conv_block1 = ConvBlock(in_channels=4, out_channels=64)
-        else:
-            self.conv_block1 = ConvBlock(in_channels=1, out_channels=64)
-        self.conv_block2 = ConvBlock(in_channels=64, out_channels=128)
-        self.conv_block3 = ConvBlock(in_channels=128, out_channels=256)
-        self.conv_block4 = ConvBlock(in_channels=256, out_channels=512)
-        self.conv_block5 = ConvBlock(in_channels=512, out_channels=1024)
-        self.conv_block6 = ConvBlock(in_channels=1024, out_channels=2048)
-
-        self.fc1 = nn.Linear(2048, 2048, bias=True)
-        self.fc_audioset = nn.Linear(2048, classes_num, bias=True)
-
-        if (self.enable_fusion) and (
-            self.fusion_type in ["daf_1d", "aff_1d", "iaff_1d"]
-        ):
-            self.mel_conv1d = nn.Sequential(
-                nn.Conv1d(64, 64, kernel_size=5, stride=3, padding=2),
-                nn.BatchNorm1d(64),  # No Relu
-            )
-            if self.fusion_type == "daf_1d":
-                self.fusion_model = DAF()
-            elif self.fusion_type == "aff_1d":
-                self.fusion_model = AFF(channels=64, type="1D")
-            elif self.fusion_type == "iaff_1d":
-                self.fusion_model = iAFF(channels=64, type="1D")
-
-        if (self.enable_fusion) and (
-            self.fusion_type in ["daf_2d", "aff_2d", "iaff_2d"]
-        ):
-            self.mel_conv2d = nn.Sequential(
-                nn.Conv2d(1, 64, kernel_size=(5, 5), stride=(6, 2), padding=(2, 2)),
-                nn.BatchNorm2d(64),
-                nn.ReLU(inplace=True),
-            )
-
-            if self.fusion_type == "daf_2d":
-                self.fusion_model = DAF()
-            elif self.fusion_type == "aff_2d":
-                self.fusion_model = AFF(channels=64, type="2D")
-            elif self.fusion_type == "iaff_2d":
-                self.fusion_model = iAFF(channels=64, type="2D")
-        self.init_weight()
-
-    def init_weight(self):
-        init_bn(self.bn0)
-        init_layer(self.fc1)
-        init_layer(self.fc_audioset)
-
-    def forward(self, input, mixup_lambda=None, device=None):
-        """
-        Input: (batch_size, data_length)"""
-
-        if self.enable_fusion and input["longer"].sum() == 0:
-            # if no audio is longer than 10s, then randomly select one audio to be longer
-            input["longer"][torch.randint(0, input["longer"].shape[0], (1,))] = True
-
-        if not self.enable_fusion:
-            x = self.spectrogram_extractor(
-                input["waveform"].to(device=device, non_blocking=True)
-            )  # (batch_size, 1, time_steps, freq_bins)
-            x = self.logmel_extractor(x)  # (batch_size, 1, time_steps, mel_bins)
-
-            x = x.transpose(1, 3)
-            x = self.bn0(x)
-            x = x.transpose(1, 3)
-        else:
-            longer_list = input["longer"].to(device=device, non_blocking=True)
-            x = input["mel_fusion"].to(device=device, non_blocking=True)
-            longer_list_idx = torch.where(longer_list)[0]
-            x = x.transpose(1, 3)
-            x = self.bn0(x)
-            x = x.transpose(1, 3)
-            if self.fusion_type in ["daf_1d", "aff_1d", "iaff_1d"]:
-                new_x = x[:, 0:1, :, :].clone().contiguous()
-                # local processing
-                if len(longer_list_idx) > 0:
-                    fusion_x_local = x[longer_list_idx, 1:, :, :].clone().contiguous()
-                    FB, FC, FT, FF = fusion_x_local.size()
-                    fusion_x_local = fusion_x_local.view(FB * FC, FT, FF)
-                    fusion_x_local = torch.permute(
-                        fusion_x_local, (0, 2, 1)
-                    ).contiguous()
-                    fusion_x_local = self.mel_conv1d(fusion_x_local)
-                    fusion_x_local = fusion_x_local.view(
-                        FB, FC, FF, fusion_x_local.size(-1)
-                    )
-                    fusion_x_local = (
-                        torch.permute(fusion_x_local, (0, 2, 1, 3))
-                        .contiguous()
-                        .flatten(2)
-                    )
-                    if fusion_x_local.size(-1) < FT:
-                        fusion_x_local = torch.cat(
-                            [
-                                fusion_x_local,
-                                torch.zeros(
-                                    (FB, FF, FT - fusion_x_local.size(-1)),
-                                    device=device,
-                                ),
-                            ],
-                            dim=-1,
-                        )
-                    else:
-                        fusion_x_local = fusion_x_local[:, :, :FT]
-                    # 1D fusion
-                    new_x = new_x.squeeze(1).permute((0, 2, 1)).contiguous()
-                    new_x[longer_list_idx] = self.fusion_model(
-                        new_x[longer_list_idx], fusion_x_local
-                    )
-                    x = new_x.permute((0, 2, 1)).contiguous()[:, None, :, :]
-                else:
-                    x = new_x
-            elif self.fusion_type in ["daf_2d", "aff_2d", "iaff_2d", "channel_map"]:
-                x = x  # no change
-
-        if self.training:
-            x = self.spec_augmenter(x)
-        # Mixup on spectrogram
-        if self.training and mixup_lambda is not None:
-            x = do_mixup(x, mixup_lambda)
-        if (self.enable_fusion) and (
-            self.fusion_type in ["daf_2d", "aff_2d", "iaff_2d"]
-        ):
-            global_x = x[:, 0:1, :, :]
-
-            # global processing
-            B, C, H, W = global_x.shape
-            global_x = self.conv_block1(global_x, pool_size=(2, 2), pool_type="avg")
-            if len(longer_list_idx) > 0:
-                local_x = x[longer_list_idx, 1:, :, :].contiguous()
-                TH = global_x.size(-2)
-                # local processing
-                B, C, H, W = local_x.shape
-                local_x = local_x.view(B * C, 1, H, W)
-                local_x = self.mel_conv2d(local_x)
-                local_x = local_x.view(
-                    B, C, local_x.size(1), local_x.size(2), local_x.size(3)
-                )
-                local_x = local_x.permute((0, 2, 1, 3, 4)).contiguous().flatten(2, 3)
-                TB, TC, _, TW = local_x.size()
-                if local_x.size(-2) < TH:
-                    local_x = torch.cat(
-                        [
-                            local_x,
-                            torch.zeros(
-                                (TB, TC, TH - local_x.size(-2), TW),
-                                device=global_x.device,
-                            ),
-                        ],
-                        dim=-2,
-                    )
-                else:
-                    local_x = local_x[:, :, :TH, :]
-
-                global_x[longer_list_idx] = self.fusion_model(
-                    global_x[longer_list_idx], local_x
-                )
-            x = global_x
-        else:
-            x = self.conv_block1(x, pool_size=(2, 2), pool_type="avg")
-
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block2(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block3(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block4(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block5(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block6(x, pool_size=(1, 1), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = torch.mean(x, dim=3)
-
-        latent_x1 = F.max_pool1d(x, kernel_size=3, stride=1, padding=1)
-        latent_x2 = F.avg_pool1d(x, kernel_size=3, stride=1, padding=1)
-        latent_x = latent_x1 + latent_x2
-        latent_x = latent_x.transpose(1, 2)
-        latent_x = F.relu_(self.fc1(latent_x))
-        latent_output = interpolate(latent_x, 32)
-
-        (x1, _) = torch.max(x, dim=2)
-        x2 = torch.mean(x, dim=2)
-        x = x1 + x2
-        x = F.dropout(x, p=0.5, training=self.training)
-        x = F.relu_(self.fc1(x))
-        embedding = F.dropout(x, p=0.5, training=self.training)
-        clipwise_output = torch.sigmoid(self.fc_audioset(x))
-
-        output_dict = {
-            "clipwise_output": clipwise_output,
-            "embedding": embedding,
-            "fine_grained_embedding": latent_output,
-        }
-        return output_dict
-
-
-class Cnn6(nn.Module):
-    def __init__(
-        self,
-        sample_rate,
-        window_size,
-        hop_size,
-        mel_bins,
-        fmin,
-        fmax,
-        classes_num,
-        enable_fusion=False,
-        fusion_type="None",
-    ):
-
-        super(Cnn6, self).__init__()
-
-        window = "hann"
-        center = True
-        pad_mode = "reflect"
-        ref = 1.0
-        amin = 1e-10
-        top_db = None
-
-        self.enable_fusion = enable_fusion
-        self.fusion_type = fusion_type
-
-        # Spectrogram extractor
-        self.spectrogram_extractor = Spectrogram(
-            n_fft=window_size,
-            hop_length=hop_size,
-            win_length=window_size,
-            window=window,
-            center=center,
-            pad_mode=pad_mode,
-            freeze_parameters=True,
-        )
-
-        # Logmel feature extractor
-        self.logmel_extractor = LogmelFilterBank(
-            sr=sample_rate,
-            n_fft=window_size,
-            n_mels=mel_bins,
-            fmin=fmin,
-            fmax=fmax,
-            ref=ref,
-            amin=amin,
-            top_db=top_db,
-            freeze_parameters=True,
-        )
-
-        # Spec augmenter
-        self.spec_augmenter = SpecAugmentation(
-            time_drop_width=64,
-            time_stripes_num=2,
-            freq_drop_width=8,
-            freq_stripes_num=2,
-        )
-
-        self.bn0 = nn.BatchNorm2d(64)
-
-        self.conv_block1 = ConvBlock5x5(in_channels=1, out_channels=64)
-        self.conv_block2 = ConvBlock5x5(in_channels=64, out_channels=128)
-        self.conv_block3 = ConvBlock5x5(in_channels=128, out_channels=256)
-        self.conv_block4 = ConvBlock5x5(in_channels=256, out_channels=512)
-
-        self.fc1 = nn.Linear(512, 512, bias=True)
-        self.fc_audioset = nn.Linear(512, classes_num, bias=True)
-
-        self.init_weight()
-
-    def init_weight(self):
-        init_bn(self.bn0)
-        init_layer(self.fc1)
-        init_layer(self.fc_audioset)
-
-    def forward(self, input, mixup_lambda=None, device=None):
-        """
-        Input: (batch_size, data_length)"""
-
-        x = self.spectrogram_extractor(input)  # (batch_size, 1, time_steps, freq_bins)
-        x = self.logmel_extractor(x)  # (batch_size, 1, time_steps, mel_bins)
-
-        x = x.transpose(1, 3)
-        x = self.bn0(x)
-        x = x.transpose(1, 3)
-
-        if self.training:
-            x = self.spec_augmenter(x)
-
-        # Mixup on spectrogram
-        if self.training and mixup_lambda is not None:
-            x = do_mixup(x, mixup_lambda)
-
-        x = self.conv_block1(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block2(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block3(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block4(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = torch.mean(x, dim=3)
-
-        latent_x1 = F.max_pool1d(x, kernel_size=3, stride=1, padding=1)
-        latent_x2 = F.avg_pool1d(x, kernel_size=3, stride=1, padding=1)
-        latent_x = latent_x1 + latent_x2
-        latent_x = latent_x.transpose(1, 2)
-        latent_x = F.relu_(self.fc1(latent_x))
-        latent_output = interpolate(latent_x, 16)
-
-        (x1, _) = torch.max(x, dim=2)
-        x2 = torch.mean(x, dim=2)
-        x = x1 + x2
-        x = F.dropout(x, p=0.5, training=self.training)
-        x = F.relu_(self.fc1(x))
-        embedding = F.dropout(x, p=0.5, training=self.training)
-        clipwise_output = torch.sigmoid(self.fc_audioset(x))
-
-        output_dict = {
-            "clipwise_output": clipwise_output,
-            "embedding": embedding,
-            "fine_grained_embedding": latent_output,
-        }
-
-        return output_dict
-
-
-class Cnn10(nn.Module):
-    def __init__(
-        self,
-        sample_rate,
-        window_size,
-        hop_size,
-        mel_bins,
-        fmin,
-        fmax,
-        classes_num,
-        enable_fusion=False,
-        fusion_type="None",
-    ):
-
-        super(Cnn10, self).__init__()
-
-        window = "hann"
-        center = True
-        pad_mode = "reflect"
-        ref = 1.0
-        amin = 1e-10
-        top_db = None
-
-        self.enable_fusion = enable_fusion
-        self.fusion_type = fusion_type
-
-        # Spectrogram extractor
-        self.spectrogram_extractor = Spectrogram(
-            n_fft=window_size,
-            hop_length=hop_size,
-            win_length=window_size,
-            window=window,
-            center=center,
-            pad_mode=pad_mode,
-            freeze_parameters=True,
-        )
-
-        # Logmel feature extractor
-        self.logmel_extractor = LogmelFilterBank(
-            sr=sample_rate,
-            n_fft=window_size,
-            n_mels=mel_bins,
-            fmin=fmin,
-            fmax=fmax,
-            ref=ref,
-            amin=amin,
-            top_db=top_db,
-            freeze_parameters=True,
-        )
-
-        # Spec augmenter
-        self.spec_augmenter = SpecAugmentation(
-            time_drop_width=64,
-            time_stripes_num=2,
-            freq_drop_width=8,
-            freq_stripes_num=2,
-        )
-
-        self.bn0 = nn.BatchNorm2d(64)
-
-        self.conv_block1 = ConvBlock(in_channels=1, out_channels=64)
-        self.conv_block2 = ConvBlock(in_channels=64, out_channels=128)
-        self.conv_block3 = ConvBlock(in_channels=128, out_channels=256)
-        self.conv_block4 = ConvBlock(in_channels=256, out_channels=512)
-        self.conv_block5 = ConvBlock(in_channels=512, out_channels=1024)
-
-        self.fc1 = nn.Linear(1024, 1024, bias=True)
-        self.fc_audioset = nn.Linear(1024, classes_num, bias=True)
-
-        self.init_weight()
-
-    def init_weight(self):
-        init_bn(self.bn0)
-        init_layer(self.fc1)
-        init_layer(self.fc_audioset)
-
-    def forward(self, input, mixup_lambda=None, device=None):
-        """
-        Input: (batch_size, data_length)"""
-
-        x = self.spectrogram_extractor(input)  # (batch_size, 1, time_steps, freq_bins)
-        x = self.logmel_extractor(x)  # (batch_size, 1, time_steps, mel_bins)
-
-        x = x.transpose(1, 3)
-        x = self.bn0(x)
-        x = x.transpose(1, 3)
-
-        if self.training:
-            x = self.spec_augmenter(x)
-
-        # Mixup on spectrogram
-        if self.training and mixup_lambda is not None:
-            x = do_mixup(x, mixup_lambda)
-
-        x = self.conv_block1(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block2(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block3(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block4(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block5(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = torch.mean(x, dim=3)
-
-        latent_x1 = F.max_pool1d(x, kernel_size=3, stride=1, padding=1)
-        latent_x2 = F.avg_pool1d(x, kernel_size=3, stride=1, padding=1)
-        latent_x = latent_x1 + latent_x2
-        latent_x = latent_x.transpose(1, 2)
-        latent_x = F.relu_(self.fc1(latent_x))
-        latent_output = interpolate(latent_x, 32)
-
-        (x1, _) = torch.max(x, dim=2)
-        x2 = torch.mean(x, dim=2)
-        x = x1 + x2
-        x = F.dropout(x, p=0.5, training=self.training)
-        x = F.relu_(self.fc1(x))
-        embedding = F.dropout(x, p=0.5, training=self.training)
-        clipwise_output = torch.sigmoid(self.fc_audioset(x))
-
-        output_dict = {
-            "clipwise_output": clipwise_output,
-            "embedding": embedding,
-            "fine_grained_embedding": latent_output,
-        }
-
-        return output_dict
-
-
-def create_pann_model(audio_cfg, enable_fusion=False, fusion_type="None"):
-    try:
-        ModelProto = eval(audio_cfg.model_name)
-        model = ModelProto(
-            sample_rate=audio_cfg.sample_rate,
-            window_size=audio_cfg.window_size,
-            hop_size=audio_cfg.hop_size,
-            mel_bins=audio_cfg.mel_bins,
-            fmin=audio_cfg.fmin,
-            fmax=audio_cfg.fmax,
-            classes_num=audio_cfg.class_num,
-            enable_fusion=enable_fusion,
-            fusion_type=fusion_type,
-        )
-        return model
-    except:
-        raise RuntimeError(
-            f"Import Model for {audio_cfg.model_name} not found, or the audio cfg parameters are not enough."
-        )

models/CLAP/open_clip/timm_model.py

@@ -1,112 +0,0 @@
-""" timm model adapter
-
-Wraps timm (https://github.com/rwightman/pytorch-image-models) models for use as a vision tower in CLIP model.
-"""
-from collections import OrderedDict
-
-import torch.nn as nn
-
-try:
-    import timm
-    from timm.models.layers import Mlp, to_2tuple
-    from timm.models.layers.attention_pool2d import RotAttentionPool2d
-    from timm.models.layers.attention_pool2d import (
-        AttentionPool2d as AbsAttentionPool2d,
-    )
-except ImportError as e:
-    timm = None
-
-from .utils import freeze_batch_norm_2d
-
-
-class TimmModel(nn.Module):
-    """timm model adapter
-    # FIXME this adapter is a work in progress, may change in ways that break weight compat
-    """
-
-    def __init__(
-        self,
-        model_name,
-        embed_dim,
-        image_size=224,
-        pool="avg",
-        proj="linear",
-        drop=0.0,
-        pretrained=False,
-    ):
-        super().__init__()
-        if timm is None:
-            raise RuntimeError("Please `pip install timm` to use timm models.")
-
-        self.image_size = to_2tuple(image_size)
-        self.trunk = timm.create_model(model_name, pretrained=pretrained)
-        feat_size = self.trunk.default_cfg.get("pool_size", None)
-        feature_ndim = 1 if not feat_size else 2
-        if pool in ("abs_attn", "rot_attn"):
-            assert feature_ndim == 2
-            # if attn pooling used, remove both classifier and default pool
-            self.trunk.reset_classifier(0, global_pool="")
-        else:
-            # reset global pool if pool config set, otherwise leave as network default
-            reset_kwargs = dict(global_pool=pool) if pool else {}
-            self.trunk.reset_classifier(0, **reset_kwargs)
-        prev_chs = self.trunk.num_features
-
-        head_layers = OrderedDict()
-        if pool == "abs_attn":
-            head_layers["pool"] = AbsAttentionPool2d(
-                prev_chs, feat_size=feat_size, out_features=embed_dim
-            )
-            prev_chs = embed_dim
-        elif pool == "rot_attn":
-            head_layers["pool"] = RotAttentionPool2d(prev_chs, out_features=embed_dim)
-            prev_chs = embed_dim
-        else:
-            assert proj, "projection layer needed if non-attention pooling is used."
-
-        # NOTE attention pool ends with a projection layer, so proj should usually be set to '' if such pooling is used
-        if proj == "linear":
-            head_layers["drop"] = nn.Dropout(drop)
-            head_layers["proj"] = nn.Linear(prev_chs, embed_dim)
-        elif proj == "mlp":
-            head_layers["mlp"] = Mlp(prev_chs, 2 * embed_dim, embed_dim, drop=drop)
-
-        self.head = nn.Sequential(head_layers)
-
-    def lock(self, unlocked_groups=0, freeze_bn_stats=False):
-        """lock modules
-        Args:
-            unlocked_groups (int): leave last n layer groups unlocked (default: 0)
-        """
-        if not unlocked_groups:
-            # lock full model
-            for param in self.trunk.parameters():
-                param.requires_grad = False
-            if freeze_bn_stats:
-                freeze_batch_norm_2d(self.trunk)
-        else:
-            # NOTE: partial freeze requires latest timm (master) branch and is subject to change
-            try:
-                # FIXME import here until API stable and in an official release
-                from timm.models.helpers import group_parameters, group_modules
-            except ImportError:
-                raise RuntimeError(
-                    "Please install latest timm `pip install git+https://github.com/rwightman/pytorch-image-models`"
-                )
-            matcher = self.trunk.group_matcher()
-            gparams = group_parameters(self.trunk, matcher)
-            max_layer_id = max(gparams.keys())
-            max_layer_id = max_layer_id - unlocked_groups
-            for group_idx in range(max_layer_id + 1):
-                group = gparams[group_idx]
-                for param in group:
-                    self.trunk.get_parameter(param).requires_grad = False
-            if freeze_bn_stats:
-                gmodules = group_modules(self.trunk, matcher, reverse=True)
-                gmodules = {k for k, v in gmodules.items() if v <= max_layer_id}
-                freeze_batch_norm_2d(self.trunk, gmodules)
-
-    def forward(self, x):
-        x = self.trunk(x)
-        x = self.head(x)
-        return x