Layers/EncoderLayer.py

# Copyright 2020 Johns Hopkins University (Shinji Watanabe)
#                Northwestern Polytechnical University (Pengcheng Guo)
#  Apache 2.0  (http://www.apache.org/licenses/LICENSE-2.0)
# Adapted by Florian Lux 2021


import torch
from torch import nn
import os
import numpy as np
from Layers.LayerNorm import LayerNorm
from Layers.Attention import MultiHeadedAttention
import matplotlib.pyplot as plt
import torch.nn.functional as F

class EncoderLayer(nn.Module):
    """
    Encoder layer module.

    Args:
        size (int): Input dimension.
        self_attn (torch.nn.Module): Self-attention module instance.
            `MultiHeadedAttention` or `RelPositionMultiHeadedAttention` instance
            can be used as the argument.
        feed_forward (torch.nn.Module): Feed-forward module instance.
            `PositionwiseFeedForward`, `MultiLayeredConv1d`, or `Conv1dLinear` instance
            can be used as the argument.
        feed_forward_macaron (torch.nn.Module): Additional feed-forward module instance.
            `PositionwiseFeedForward`, `MultiLayeredConv1d`, or `Conv1dLinear` instance
            can be used as the argument.
        conv_module (torch.nn.Module): Convolution module instance.
            `ConvlutionModule` instance can be used as the argument.
        dropout_rate (float): Dropout rate.
        normalize_before (bool): Whether to use layer_norm before the first block.
        concat_after (bool): Whether to concat attention layer's input and output.
            if True, additional linear will be applied.
            i.e. x -> x + linear(concat(x, att(x)))
            if False, no additional linear will be applied. i.e. x -> x + att(x)

    """

    def __init__(self, size, self_attn, feed_forward, feed_forward_macaron, conv_module, dropout_rate, normalize_before=True, concat_after=False, ):
        super(EncoderLayer, self).__init__()
        self.self_attn = self_attn
        self.feed_forward = feed_forward
        self.feed_forward_macaron = feed_forward_macaron
        self.conv_module = conv_module
        self.norm_ff = LayerNorm(size)  # for the FNN module
        self.norm_mha = LayerNorm(size)  # for the MHA module
        if feed_forward_macaron is not None:
            self.norm_ff_macaron = LayerNorm(size)
            self.ff_scale = 0.5
        else:
            self.ff_scale = 1.0
        if self.conv_module is not None:
            self.norm_conv = LayerNorm(size)  # for the CNN module
            self.norm_final = LayerNorm(size)  # for the final output of the block
        self.dropout = nn.Dropout(dropout_rate)
        self.size = size
        self.normalize_before = normalize_before
        self.concat_after = concat_after
        if self.concat_after:
            self.concat_linear = nn.Linear(size + size, size)

    def forward(self, x_input, mask, cache=None):
        """
        Compute encoded features.

        Args:
            x_input (Union[Tuple, torch.Tensor]): Input tensor w/ or w/o pos emb.
                - w/ pos emb: Tuple of tensors [(#batch, time, size), (1, time, size)].
                - w/o pos emb: Tensor (#batch, time, size).
            mask (torch.Tensor): Mask tensor for the input (#batch, time).
            cache (torch.Tensor): Cache tensor of the input (#batch, time - 1, size).

        Returns:
            torch.Tensor: Output tensor (#batch, time, size).
            torch.Tensor: Mask tensor (#batch, time).

        """
        if isinstance(x_input, tuple):
            x, pos_emb = x_input[0], x_input[1]
        else:
            x, pos_emb = x_input, None

        # whether to use macaron style
        if self.feed_forward_macaron is not None:
            residual = x
            if self.normalize_before:
                x = self.norm_ff_macaron(x)
            x = residual + self.ff_scale * self.dropout(self.feed_forward_macaron(x))
            if not self.normalize_before:
                x = self.norm_ff_macaron(x)

        # multi-headed self-attention module
        residual = x
        if self.normalize_before:
            x = self.norm_mha(x)

        if cache is None:
            x_q = x
        else:
            assert cache.shape == (x.shape[0], x.shape[1] - 1, self.size)
            x_q = x[:, -1:, :]
            residual = residual[:, -1:, :]
            mask = None if mask is None else mask[:, -1:, :]

        if pos_emb is not None:
            x_att = self.self_attn(x_q, x, x, pos_emb, mask)
        else:
            x_att = self.self_attn(x_q, x, x, mask)

        if self.concat_after:
            x_concat = torch.cat((x, x_att), dim=-1)
            x = residual + self.concat_linear(x_concat)
        else:
            x = residual + self.dropout(x_att)
        if not self.normalize_before:
            x = self.norm_mha(x)

        # convolution module
        if self.conv_module is not None:
            residual = x
            if self.normalize_before:
                x = self.norm_conv(x)
            x = residual + self.dropout(self.conv_module(x))
            if not self.normalize_before:
                x = self.norm_conv(x)

        # feed forward module
        residual = x
        if self.normalize_before:
            x = self.norm_ff(x)
        x = residual + self.ff_scale * self.dropout(self.feed_forward(x))
        if not self.normalize_before:
            x = self.norm_ff(x)

        if self.conv_module is not None:
            x = self.norm_final(x)

        if cache is not None:
            x = torch.cat([cache, x], dim=1)


#####################
        # x_q = x.to(device='cuda:0')
        # batch = x_q.shape[0] # save batch size as it is query
        # residual = x
        # self.accent_emb = []
        # for embs in next(os.walk('/nas/projects/vokquant/IMS-Toucan_lang_emb_conformer/Preprocessing/embeds/'))[2]:
        #     self.accent_emb.append(torch.load(os.path.join('/nas/projects/vokquant/IMS-Toucan_lang_emb_conformer/Preprocessing/embeds/', embs )))
        # self.accent_emb = np.concatenate(self.accent_emb)       # created shape: (12, 1, 192)
        # self.accent_emb = torch.Tensor(self.accent_emb).repeat(1,1,1,2)
        # self.accent_emb = self.accent_emb.squeeze(0)
        # x_acc_emb = self.accent_emb.permute(1, 0, 2).repeat(batch,1,1).to(device='cuda:0')
        
        # # check if x_acc_emb and x_q is in cpu or gpu
        # #print("x_acc_emb.device: ", x_acc_emb.device, "x_q.device: ", x_q.device)
        # #print("x_acc_emb.shape: ", x_acc_emb.shape, "x_q.shape: ", x_q.shape)
        # #self.self_attn = MultiHeadedAttention(4, 384, 0.1).to(device='cuda:0') # produces dimension error
        # multihead_attn = nn.MultiheadAttention(384, 16, 0.1, batch_first=True).to(device='cuda:0')
        # attn_output, attn_weights = multihead_attn(x_q, x_acc_emb, x_acc_emb)
        # print("attn_weights.shape: ", attn_weights.shape)
        # # in inference
        # # make all other speakers zeros
        # #x_acc_emb[:,1:,:] = 0

        # # visualize attn_weights in a plot
        # # assuming attn_weights has shape (batch_size, query_length, key_length)
        # attn_weights_0 = attn_weights[0].to(device='cpu')

        # # apply softmax to get values between 0 and 1 that sum to 1
        # attn_weights_softmax_0 = F.softmax(attn_weights_0, dim=1)
        
        # # plot the attention weights as a heatmap
        # plt.imshow(attn_weights_softmax_0.detach().numpy(), cmap='hot', interpolation='nearest')
        # plt.colorbar()
        # plt.savefig('heatmap.png')

        # #print("attn_output.shape: ", attn_output.shape, "attn_weights.shape: ", attn_weights.shape, "\n")
        # x = residual + self.dropout(attn_output)
        # x = self.norm_mha(x)
        # # print("pos_emb.shape: ", pos_emb.shape)
        # # print("mask.shape: ", mask.shape)
        # # x = self.self_attn(x_q, x_acc_emb, x_acc_emb, pos_emb)

        # # side info: it appears, that https://pytorch.org/docs/stable/_modules/torch/nn/modules/activation.html#MultiheadAttention.forward does a call to F.multi_head_attention_forward(... which in github shows as: https://github.com/pytorch/pytorch/blob/8372c5dc687d622c7d2e0d411f61cd2720fc1052/torch/nn/functional.py#L5029 see:
        # # scale (optional float): Scaling factor applied prior to softmax. If None, the default value is set to :math:`\frac{1}{\sqrt{E}}`.
        # # where E is the embedding dimension. So it should be the same as in the paper SPEAKER-AWARE SPEECH-TRANSFORMER (formula 7)

#####################
        if pos_emb is not None:
            return (x, pos_emb), mask
        return x, mask