"""code in this module adapted from: https://github.com/juho-lee/set_transformer/blob/master/modules.py"""
from typing import Any

import torch
import torch.nn as nn
import torch.nn.functional as F
import math
import pytorch_lightning as pl


class MAB(pl.LightningModule):
    def __init__(self, dim_Q, dim_K, dim_V, num_heads, ln=False):
        super(MAB, self).__init__()
        self.dim_V = dim_V
        self.num_heads = num_heads
        self.fc_q = nn.Linear(dim_Q, dim_V)
        self.fc_k = nn.Linear(dim_K, dim_V)
        self.fc_v = nn.Linear(dim_K, dim_V)
        if ln:
            self.ln0 = nn.LayerNorm(dim_V)
            self.ln1 = nn.LayerNorm(dim_V)
        self.fc_o = nn.Linear(dim_V, dim_V)

    def forward(self, Q, K, mask_k=None, mask_q=None):
        Q = self.fc_q(Q)
        K, V = self.fc_k(K), self.fc_v(K)

        dim_split = self.dim_V // self.num_heads
        Q_ = torch.cat(Q.split(dim_split, 2), 0)
        K_ = torch.cat(K.split(dim_split, 2), 0)
        V_ = torch.cat(V.split(dim_split, 2), 0)

        prod = Q_.bmm(K_.transpose(1, 2)) / math.sqrt(self.dim_V)
        if mask_k is not None:
            prod[~mask_k.unsqueeze(1).repeat(self.num_heads, prod.shape[1], 1)] = float('-inf')
        A = torch.softmax(prod, 2)
        O = torch.cat((Q_ + A.bmm(V_)).split(Q.size(0), 0), 2)
        O = O.nan_to_num(0)
        O = O if getattr(self, 'ln0', None) is None else self.ln0(O)
        O = O + F.relu(self.fc_o(O))
        O = O if getattr(self, 'ln1', None) is None else self.ln1(O)
        return O


class SAB(pl.LightningModule):
    def __init__(self, dim_in, dim_out, num_heads, ln=False):
        super(SAB, self).__init__()
        self.mab = MAB(dim_in, dim_in, dim_out, num_heads, ln=ln)

    def forward(self, inputs):
        X = inputs['X']
        mask = inputs['mask']
        return {'X': self.mab(X, X, mask_k=mask, mask_q=mask), 'mask': mask}


class ISAB(pl.LightningModule):
    def __init__(self, dim_in, dim_out, num_heads, num_inds, ln=False):
        super(ISAB, self).__init__()
        self.I = nn.Parameter(torch.Tensor(1, num_inds, dim_out))
        nn.init.xavier_uniform_(self.I)
        self.mab0 = MAB(dim_out, dim_in, dim_out, num_heads, ln=ln)
        self.mab1 = MAB(dim_in, dim_out, dim_out, num_heads, ln=ln)

    def forward(self, inputs):
        X = inputs['X']
        mask = inputs['mask']
        H = self.mab0(self.I.repeat(X.size(0), 1, 1), X, mask_k=mask)
        isab_out = self.mab1(X, H, mask_q=mask)
        return {'X': isab_out, 'mask': mask}


class PMA(pl.LightningModule):
    def __init__(self, dim, num_heads, num_seeds, ln=False):
        super(PMA, self).__init__()
        self.S = nn.Parameter(torch.Tensor(1, num_seeds, dim))
        nn.init.xavier_uniform_(self.S)
        self.mab = MAB(dim, dim, dim, num_heads, ln=ln)

    def forward(self, inputs):
        X = inputs['X']
        mask = inputs['mask']
        return {'X': self.mab(self.S.repeat(X.size(0), 1, 1), X, mask_k=mask), 'mask': None}


class CustomLinear(pl.LightningModule):
    def __init__(self, dim_hidden, dim_output) -> None:
        super().__init__()
        self.linear = nn.Linear(dim_hidden, dim_output)

    def forward(self, inputs) -> Any:
        X = inputs['X']
        return self.linear(X)


class SetTransformer(pl.LightningModule):
    def __init__(self, dim_input, num_outputs, dim_output,
                 num_inds=32, dim_hidden=128, num_heads=4, ln=False):
        super(SetTransformer, self).__init__()
        self.enc = nn.Sequential(
            ISAB(dim_input, dim_hidden, num_heads, num_inds, ln=ln),
            ISAB(dim_hidden, dim_hidden, num_heads, num_inds, ln=ln))
        self.dec = nn.Sequential(
            PMA(dim_hidden, num_heads, num_outputs, ln=ln),
            SAB(dim_hidden, dim_hidden, num_heads, ln=ln),
            SAB(dim_hidden, dim_hidden, num_heads, ln=ln),
            CustomLinear(dim_hidden, dim_output))

    def forward(self, X, mask=None):
        encoded = self.enc({'X': X, 'mask': mask})
        return self.dec(encoded)

    @staticmethod
    def pad_and_mask(X):
        lens = torch.tensor([x.shape[0] for x in X])
        max_len = lens.max()
        mask = torch.arange(max_len)[None, :] < lens[:, None]
        padded = torch.nn.utils.rnn.pad_sequence(X, batch_first=True)
        return padded, mask

