<jupyter_start><jupyter_text>RNN_FCNPlus> This is an unofficial PyTorch implementation by <NAME> - <EMAIL> based on:* <NAME>., <NAME>., <NAME>., & <NAME>. (2017). LSTM fully convolutional networks for time series classification. IEEE Access, 6, 1662-1669.* Official LSTM-FCN TensorFlow implementation: https://github.com/titu1994/LSTM-FCN* <NAME>., <NAME>., & <NAME>. (2018). Deep Gated Recurrent and Convolutional Network Hybrid Model for Univariate Time Series Classification. arXiv preprint arXiv:1812.07683.* Official GRU-FCN TensorFlow implementation: https://github.com/NellyElsayed/GRU-FCN-model-for-univariate-time-series-classification<jupyter_code>#export
from tsai.imports import *
from tsai.models.layers import *
#export
class _RNN_FCN_BasePlus(nn.Sequential):
    def __init__(self, c_in, c_out, seq_len=None, hidden_size=100, rnn_layers=1, bias=True, cell_dropout=0, rnn_dropout=0.8, bidirectional=False, shuffle=True, 
                 fc_dropout=0., conv_layers=[128, 256, 128], kss=[7, 5, 3], se=0):
        
        if shuffle: assert seq_len is not None, 'need seq_len if shuffle=True'
        
        backbone = _RNN_FCN_Base_Backbone(self._cell, c_in, c_out, seq_len=seq_len, hidden_size=hidden_size, rnn_layers=rnn_layers, bias=bias, 
                                          cell_dropout=cell_dropout, rnn_dropout=rnn_dropout, bidirectional=bidirectional, shuffle=shuffle, 
                                          conv_layers=conv_layers, kss=kss, se=se)
        
        head_layers = [nn.Dropout(fc_dropout)] if fc_dropout else []
        head_layers += [nn.Linear(hidden_size * (1 + bidirectional) + conv_layers[-1], c_out)] 
        head = nn.Sequential(*head_layers)
        
        layers = OrderedDict([('backbone', backbone), ('head', head)])
        super().__init__(layers)

            
class _RNN_FCN_Base_Backbone(Module):
    def __init__(self, _cell, c_in, c_out, seq_len=None, hidden_size=100, rnn_layers=1, bias=True, cell_dropout=0, rnn_dropout=0.8, bidirectional=False, 
                 shuffle=True, conv_layers=[128, 256, 128], kss=[7, 5, 3], se=0):
        
        # RNN - first arg is usually c_in. Authors modified this to seq_len by not permuting x. This is what they call shuffled data.
        self.rnn = _cell(seq_len if shuffle else c_in, hidden_size, num_layers=rnn_layers, bias=bias, batch_first=True, 
                              dropout=cell_dropout, bidirectional=bidirectional)
        self.rnn_dropout = nn.Dropout(rnn_dropout) if rnn_dropout else noop
        self.shuffle = Permute(0,2,1) if not shuffle else noop # You would normally permute x. Authors did the opposite.
        
        # FCN
        assert len(conv_layers) == len(kss)
        self.convblock1 = ConvBlock(c_in, conv_layers[0], kss[0])
        self.se1 = SqueezeExciteBlock(conv_layers[0], se) if se != 0 else noop
        self.convblock2 = ConvBlock(conv_layers[0], conv_layers[1], kss[1])
        self.se2 = SqueezeExciteBlock(conv_layers[1], se) if se != 0 else noop
        self.convblock3 = ConvBlock(conv_layers[1], conv_layers[2], kss[2])
        self.gap = GAP1d(1)
        
        # Common
        self.concat = Concat()
        
    def forward(self, x):  
        # RNN
        rnn_input = self.shuffle(x) # permute --> (batch_size, seq_len, n_vars) when batch_first=True
        output, _ = self.rnn(rnn_input)
        last_out = output[:, -1] # output of last sequence step (many-to-one)
        last_out = self.rnn_dropout(last_out)
        
        # FCN
        x = self.convblock1(x)
        x = self.se1(x)
        x = self.convblock2(x)
        x = self.se2(x)
        x = self.convblock3(x)
        x = self.gap(x)

        # Concat
        x = self.concat([last_out, x])
        return x
    

class RNN_FCNPlus(_RNN_FCN_BasePlus):
    _cell = nn.RNN
    
class LSTM_FCNPlus(_RNN_FCN_BasePlus):
    _cell = nn.LSTM
    
class GRU_FCNPlus(_RNN_FCN_BasePlus):
    _cell = nn.GRU
    
class MRNN_FCNPlus(_RNN_FCN_BasePlus):
    _cell = nn.RNN
    def __init__(self, *args, se=16, **kwargs):
        super().__init__(*args, se=16, **kwargs)
    
class MLSTM_FCNPlus(_RNN_FCN_BasePlus):
    _cell = nn.LSTM
    def __init__(self, *args, se=16, **kwargs):
        super().__init__(*args, se=16, **kwargs)
    
class MGRU_FCNPlus(_RNN_FCN_BasePlus):
    _cell = nn.GRU
    def __init__(self, *args, se=16, **kwargs):
        super().__init__(*args, se=16, **kwargs)
from tsai.models.utils import count_parameters
from tsai.models.RNN_FCN import *
bs = 16
n_vars = 3
seq_len = 12
c_out = 2
xb = torch.rand(bs, n_vars, seq_len)
test_eq(RNN_FCNPlus(n_vars, c_out, seq_len)(xb).shape, [bs, c_out])
test_eq(LSTM_FCNPlus(n_vars, c_out, seq_len)(xb).shape, [bs, c_out])
test_eq(MLSTM_FCNPlus(n_vars, c_out, seq_len)(xb).shape, [bs, c_out])
test_eq(GRU_FCNPlus(n_vars, c_out, shuffle=False)(xb).shape, [bs, c_out])
test_eq(GRU_FCNPlus(n_vars, c_out, seq_len, shuffle=False)(xb).shape, [bs, c_out])
test_eq(count_parameters(LSTM_FCNPlus(n_vars, c_out, seq_len)), count_parameters(LSTM_FCN(n_vars, c_out, seq_len)))
LSTM_FCNPlus(n_vars, seq_len, c_out, se=8)
#hide
out = create_scripts(); beep(out)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12333268896191765, 'fraction_numerical': 0.023197370964624008, 'mean_word_length': 3.2937759336099584, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 16, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '125193', 'license': 'Apache-2.0', 'n_tokens_mistral': 1991, 'n_tokens_neox': 1883, 'n_words': 399, 'path': 'nbs/107b_models.RNN_FCNPlus.ipynb', 'repo_name': 'nwsrnm/tsai', 'size': '5214'}