【网络架构】Convolutional LSTM Network: A Machine Learning Approach for Precipitation Nowcasting

Convolutional LSTM Network: A Machine LearningApproach for Precipitation Nowcasting

这篇文章主要是了解方法.

原始文档: https://www.yuque.com/lart/papers/nvx1re

这篇文章主要提出了一种改进的卷积实现的LSTM结构. 从而更好的利用时空特征.

LSTM大体历史回顾

原始LSTM

圆圈是CEC, 里面是一条y = x的直线表示该神经元的激活函数是线性的，自链接的权重为1.

+ Forget Gate

+ Peehole

简化版示意图:

这里提到了参考文献2提出的一种改进LSTM结构, 较以前的LSTM的改进主要是添加了Peehole(窥视孔), 将细胞单元链接到了输入门, 遗忘门, 输出门上. 该文章中提到的FC-LSTM实际上就是这样的一类LSTM结构. 它的计算方法是:

其中的小圆圈表示哈达吗乘积, 也就是元素间的乘积运算. 能够看出来, 这里在输入门, 遗忘门, 输出门的输入上, 都考虑了细胞状态c_{t-1}, 与原始的LSTM不一样

+ Convolution

本文的想法就是使用了卷及操做来代替矩阵乘法的操做.

虽然FC-LSTM层已被证实对处理时间相关性颇有效, 但它对空间数据的处理, 包含太多冗余. FC-LSTM在处理时空数据时的主要缺点是它在输入到状态和状态到状态转换中使用全链接，其中没有空间信息被编码.

这里提出了FC-LSTM的扩展，它在输入到状态和状态到状态转换中都具备卷积结构. 经过堆叠多个ConvLSTM层并造成编码预测结构，能够创建更通常的时空序列预测模型。

文章的设计的一个显着特色是全部输入X1, ..., Xt, 细胞输出C1, ..., Ct, 隐藏状态H1, ..., Ht, 和ConvLSTM的几个门it, ft, ot是都是3维张量, 它们的最后两个维度是空间维度(行和列).

这里的对应的公式以下:

若是将状态视为移动对象的隐藏表示，具备大转换核的ConvLSTM应该可以捕获更快的运动，而具备较小核的ConvLSTM可以捕获较慢的运动。此外, 前面FC-LSTM公式表示的输入, 细胞输出和隐藏状态, 也能够被视为3维张量. 只是它们最后两个维度为1. 在这个意义上, FC-LSTM其实是ConvLSTM的一个特例, 其中全部特征都"站"在一个单元格上.

• 为了确保状态具备与输入相同的行数和相同的列数，在应用卷积运算以前须要padding。这里边界点上隐藏状态的填充能够被视为使用外部世界的状态进行计算。
• 一般，在第一个输入到来以前，将LSTM的全部状态初始化为零，这对应于对于将来的“彻底无知”。

相似地，若是对隐藏状态执行零填充（在本文中使用），实际上将外部世界的状态设置为零而且假设没有关于外部的预知。经过填充状态，能够区别对待边界点，这在许多状况下是有帮助的。例如，假设观察的系统是被墙围绕的移动球。虽然看不到这些墙，但们能够经过一次又一次地发现球在它们上面弹跳来推断它们的存在，若是边界点具备与内点相同的状态转移动力学(the same state transition dynamics)，则很难作到这一点。

编解码结构

与FC-LSTM同样，ConvLSTM也能够做为更复杂结构的构建块。对于咱们的时空序列预测问题，咱们使用图3所示的结构，它包括两个网络，一个编码网络和一个预测网络。预测网络的初始状态和单元输出是从编码网络的最后状态复制的。两个网络都是经过堆叠多个ConvLSTM层造成的。

因为咱们的预测目标与输入具备相同的维度，咱们将预测网络中的全部状态链接起来并将它们馈送到1x1卷积层以生成最终预测。

代码启发

这里代码的实现, 让我学习到了对于LSTM处理图片类的数据的时候, (时空)计算的特殊之处. 时间步 和不一样 ConvLSTMCell 的堆叠之间, 有关联有分离. 同一时间步内, 会存在多个Cell的堆叠计算, 而只输入一次原始数据, 而且, 每个Cell的输出都会做为下一时间步的输入, 同时, 在下一时间步里, 原始输入仍是同样的. 总体时间步展开, 构成了一个网格状的结构. 关键的一点是, 每一个时间步对应的Cell的卷积权重是一致的. 由于使用的是相同的卷积层.

self._all_layers = []
        for i in range(self.num_layers):
            name = 'cell{}'.format(i)
            cell = ConvLSTMCell(self.input_channels[i],
                                self.hidden_channels[i],
                                self.kernel_size,
                                self.bias)
            # 设定 self.cell{i} = cell 很好的方法, 值得借鉴, 后期添加属性
            setattr(self, name, cell)
            self._all_layers.append(cell)

大体手绘了一下时间步为5, 每一个时间步有5个Cell的展开结构:

代码参考

import torch
import torch.nn as nn


class ConvLSTMCell(nn.Module):
    def __init__(self, input_channels, hidden_channels, kernel_size, bias=True):
        super(ConvLSTMCell, self).__init__()

        assert hidden_channels % 2 == 0

        self.input_channels = input_channels
        self.hidden_channels = hidden_channels
        self.bias = bias
        self.kernel_size = kernel_size
        self.num_features = 4

        # N=(W?F+2P)/S+1
        self.padding = int((kernel_size - 1) / 2)

        self.Wxi = nn.Conv2d(self.input_channels, self.hidden_channels,
                             self.kernel_size, 1, self.padding, bias=True)
        self.Whi = nn.Conv2d(self.hidden_channels, self.hidden_channels,
                             self.kernel_size, 1, self.padding, bias=False)

        self.Wxf = nn.Conv2d(self.input_channels, self.hidden_channels,
                             self.kernel_size, 1, self.padding, bias=True)
        self.Whf = nn.Conv2d(self.hidden_channels, self.hidden_channels,
                             self.kernel_size, 1, self.padding, bias=False)

        self.Wxc = nn.Conv2d(self.input_channels, self.hidden_channels,
                             self.kernel_size, 1, self.padding, bias=True)
        self.Whc = nn.Conv2d(self.hidden_channels, self.hidden_channels,
                             self.kernel_size, 1, self.padding, bias=False)

        self.Wxo = nn.Conv2d(self.input_channels, self.hidden_channels,
                             self.kernel_size, 1, self.padding, bias=True)
        self.Who = nn.Conv2d(self.hidden_channels, self.hidden_channels,
                             self.kernel_size, 1, self.padding, bias=False)

        self.Wci = None
        self.Wcf = None
        self.Wco = None

    def forward(self, x, h, c):
        ci = torch.sigmoid(self.Wxi(x) + self.Whi(h) + c * self.Wci)
        cf = torch.sigmoid(self.Wxf(x) + self.Whf(h) + c * self.Wcf)
        cc = cf * c + ci * torch.tanh(self.Wxc(x) + self.Whc(h))
        co = torch.sigmoid(self.Wxo(x) + self.Who(h) + cc * self.Wco)
        ch = co * torch.tanh(cc)
        return ch, cc

    def init_hidden(self, batch_size, hidden, shape):
        self.Wci = torch.zeros(1, hidden, shape[0], shape[1]).cuda()
        self.Wcf = torch.zeros(1, hidden, shape[0], shape[1]).cuda()
        self.Wco = torch.zeros(1, hidden, shape[0], shape[1]).cuda()
        return torch.zeros(batch_size, hidden, shape[0], shape[1]).cuda(), \
               torch.zeros(batch_size, hidden, shape[0], shape[1]).cuda()


class ConvLSTM(nn.Module):
    # input_channels corresponds to the first input feature map
    # hidden state is a list of succeeding lstm layers.
    def __init__(self,
                 input_channels,
                 hidden_channels,
                 kernel_size,
                 step=2,
                 effective_step=[1],
                 bias=True):
        """
        :param input_channels: 输入通道数
        :param hidden_channels: 隐藏通道数, 是个列表, 能够表示这个ConvLSTM内部每一层结构
        :param kernel_size: 卷积实现对应的核尺寸
        :param step: 该ConvLSTM自身总的循环次数
        :param effective_step: 输出中将要使用的步数(不必定全用)
        :param bias: 各个门的偏置项
        """
        super(ConvLSTM, self).__init__()

        self.input_channels = [input_channels] + hidden_channels
        self.hidden_channels = hidden_channels
        self.kernel_size = kernel_size
        self.num_layers = len(hidden_channels)
        self.step = step
        self.bias = bias
        self.effective_step = effective_step

        self._all_layers = []
        for i in range(self.num_layers):
            name = 'cell{}'.format(i)
            cell = ConvLSTMCell(self.input_channels[i],
                                self.hidden_channels[i],
                                self.kernel_size,
                                self.bias)
            # 设定 self.cell{i} = cell 很好的方法, 值得借鉴, 后期添加属性
            setattr(self, name, cell)
            self._all_layers.append(cell)

    def forward(self, input):
        internal_state = []
        outputs = []
        for step in range(self.step):
            """
            每一个时间步里都要进行对原始输入`input`的多个ConvLSTMCell的的级联处理.
            而第一个时间步里, 设定各个ConvLSTMCell全部的初始h与c都是0.
            各个ConvLSTMCell的输出h和c都是下一个时间步下对应的ConvLSTMCell的输入用的h和c, 
            各个ConvLSTMCell的输入都是同一时间步下上一个ConvLSTMCell的输出的h(做为input项)
            和自身对应的h和c.
            """
            x = input

            # 对每种隐藏状态尺寸来进行叠加
            for i in range(self.num_layers):
                # all cells are initialized in the first step
                name = f'cell{i}'

                # 初始化各个ConvLSTM的门里的Peehole权重为0
                if step == 0:
                    bsize, _, height, width = x.size()

                    # getattr得到了对应的self.cell{i}的值, 也就是对应的层
                    (h, c) = getattr(self, name).init_hidden(
                        batch_size=bsize,
                        hidden=self.hidden_channels[i],
                        shape=(height, width)
                    )
                    # 第一步里的h和c都是0
                    internal_state.append((h, c))

                # do forward
                (h, c) = internal_state[i]
                x, new_c = getattr(self, name)(x, h, c)
                # update new h&c
                internal_state[i] = (x, new_c)
            # only record effective steps
            if step in self.effective_step:
                outputs.append(x)

        return outputs, (x, new_c)

使用方法:

if __name__ == '__main__':
    # gradient check
    convlstm = ConvLSTM(input_channels=512,
                        hidden_channels=[128, 64, 64, 32, 32],
                        kernel_size=3,
                        step=2,  # 这里最后会断定有效的步的输出, 要断定是否step in eff_steps, 因此得保证step可能在列表中
                        effective_step=[1]).cuda()
    loss_fn = torch.nn.MSELoss()

    input = torch.randn(1, 512, 64, 32).cuda()
    target = torch.randn(1, 32, 64, 32, requires_grad=True, dtype=torch.float64).cuda()

    output, (x, new_c) = convlstm(input)
    print(output[0].size())
    output = output[0].double()
    res = torch.autograd.gradcheck(loss_fn,
                                   (output, target),
                                   eps=1e-6,
                                   raise_exception=True)
    print(res)
    
 
# 输出
# torch.Size([1, 32, 64, 32])
# True

参考文章

1. Generating Sequences With Recurrent Neural Networks
2. 关于Peehole的改进的提出: https://www.researchgate.net/publication/2562741_Long_Short-Term_Memory_in_Recurrent_Neural_Networks?enrichId=rgreq-8d9f795da6b29cae037bf9e0cb943d7a-XXX&enrichSource=Y292ZXJQYWdlOzI1NjI3NDE7QVM6MzcxMDEwMjU2ODE4MTc2QDE0NjU0NjcxNDYwMjU%3D&el=1_x_3&_esc=publicationCoverPdf
3. https://blog.csdn.net/xmdxcsj/article/details/52526843
4. http://www.javashuo.com/article/p-ygoieree-md.html
5. https://blog.csdn.net/sinat_26917383/article/details/71817742
6. 文中代码来自: https://github.com/automan000/Convolution_LSTM_PyTorch