【網絡架構】Convolutional LSTM Network: A Machine Learning Approach for Precipitation Nowcasting

時間 2019-11-10

標籤網絡架構 convolutional lstm network machine learning approach precipitation nowcasting 欄目系統網絡简体版

原文原文鏈接

Convolutional LSTM Network: A Machine LearningApproach for Precipitation Nowcasting

這篇文章主要是瞭解方法.python

原始文檔: https://www.yuque.com/lart/papers/nvx1regit

這篇文章主要提出了一種改進的卷積實現的LSTM結構. 從而更好的利用時空特徵.github

LSTM大體歷史回顧

原始LSTM

圓圈是CEC, 裏面是一條y = x的直線表示該神經元的激活函數是線性的，自鏈接的權重爲1.網絡

+ Forget Gate

+ Peehole

簡化版示意圖:app

這裏提到了參考文獻2提出的一種改進LSTM結構, 較以前的LSTM的改進主要是添加了Peehole(窺視孔), 將細胞單元鏈接到了輸入門, 遺忘門, 輸出門上. 該文章中提到的FC-LSTM實際上就是這樣的一類LSTM結構. 它的計算方法是:函數

其中的小圓圈表示哈達嗎乘積, 也就是元素間的乘積運算. 能夠看出來, 這裏在輸入門, 遺忘門, 輸出門的輸入上, 都考慮了細胞狀態c_{t-1}, 與原始的LSTM不一樣學習

+ Convolution

本文的想法就是使用了卷及操做來代替矩陣乘法的操做.ui

雖然FC-LSTM層已被證實對處理時間相關性頗有效, 但它對空間數據的處理, 包含太多冗餘. FC-LSTM在處理時空數據時的主要缺點是它在輸入到狀態和狀態到狀態轉換中使用全鏈接，其中沒有空間信息被編碼.編碼

這裏提出了FC-LSTM的擴展，它在輸入到狀態和狀態到狀態轉換中都具備卷積結構. 經過堆疊多個ConvLSTM層並造成編碼預測結構，能夠創建更通常的時空序列預測模型。.net

文章的設計的一個顯着特色是全部輸入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