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. https://blog.csdn.net/shincling/article/details/49362161
5. https://blog.csdn.net/sinat_26917383/article/details/71817742
6. 文中代碼來自: https://github.com/automan000/Convolution_LSTM_PyTorch

Convolutional LSTM Network: A Machine LearningApproach for Precipitation Nowcasting