주요 모듈 import 하기
import torch
import torch.nn as nn
from torch.nn.utils import weight_norm
normalization 을 weight normalization 으로 하기 때문에 해당하는 모듈도 같이 가져옵니다.
chomp 1d
chomp 가 필요한 이유는 파이토치에서 패딩을 줄 때 좌 우를 모두 주기 때문입니다.
class Chomp1d(nn.Module):
def __init__(self, chomp_size):
super(Chomp1d, self).__init__()
self.chomp_size = chomp_size
def forward(self, x):
return x[:, :, :-self.chomp_size].contiguous()
파이토치에서 conv1d 에 (batch size, # features, input length) 를 인풋으로 넣으면 (batch size, # channels, output length) shape 이 output 으로 출력됩니다.
따라서 뒤에 붙은 패딩으로 원하던 output length (=input length) 보다 길어진 부분은 chomp 해줍니다. ppt 에서는 제가 케라스 conv1d 를 기준으로 적어서 (batch size, input length, output size) 로 표시되었습니다.
파이토치에서 자주 보이는 continguous 도 이 참에 직관적이게나마 정리해보았습니다.
친절한 pytorch kr 에서 ‘contiguous함수는 텐서를 numpy같은 방식으로 메모리에 저장하는 방식을 말합니다’ 라고 알려주셨는데
그래서 numpy 식 데이터 저장 방식과 리스트 식 데이터 저장 방식을 검색해보았습니다.
파이토치에서 reshape 에 쓰이는 view 메소드의 경우 명칭 그대로, 데이터 자체의 shape 을 바꾸는 것이 아니라 데이터 shape 은 그대로인데 그것을 다른 shape 으로 ‘view’ 하는 것이므로 ‘view함수는 reshape나 resize와는 다르게 어떤 경우에도 메모리복사없이 이루어집니다 따라서 contiguous형식이 아닐때는 텐서모양을 고칠수 없게되고 런타임에러가 발생합니다’ 이라고 알려주신 것처럼 그러면 view 할 수 있게끔 메모리를 continguous 한 형식으로 담아주어야 합니다.
TemporalBlock
###
이렇게 구성되기 때문에
class TemporalBlock(nn.Module):
def __init__(self, n_inputs, n_outputs, kernel_size, stride, dilation, padding, dropout=0.2):
super(TemporalBlock, self).__init__()
self.conv1 = weight_norm(nn.Conv1d(n_inputs, n_outputs, kernel_size,
stride=stride, padding=padding, dilation=dilation))
self.chomp1 = Chomp1d(padding)
self.relu1 = nn.ReLU()
self.dropout1 = nn.Dropout(dropout)
self.conv2 = weight_norm(nn.Conv1d(n_outputs, n_outputs, kernel_size,
stride=stride, padding=padding, dilation=dilation))
self.chomp2 = Chomp1d(padding)
self.relu2 = nn.ReLU()
self.dropout2 = nn.Dropout(dropout)
self.net = nn.Sequential(self.conv1, self.chomp1, self.relu1, self.dropout1,
self.conv2, self.chomp2, self.relu2, self.dropout2)
self.downsample = nn.Conv1d(n_inputs, n_outputs, 1) if n_inputs != n_outputs else None
self.relu = nn.ReLU()
self.init_weights()
self.conv1 은 TCN1 + weight normalization 을 나타내고, 뒤에 불필요하게 들어가서 길이 안 맞는 padding 은 chomp 로 잘라주고 (원래 뒤에 불필요한 padding 안 들어가면 output_length=input_length) relu 해주고 dropout 해주고 똑같이 한번 더 반복됩니다.
downsample 의 경우,
TCN 은 여러 cnn 레이어가 합쳐진 형태입니다. 밑에 TemporalConvNet 의 인자인 num_channels 에는 총 num_levels 개의 cnn 레이어에서 각 cnn 의 채널을 몇 개로 할지 num_levels 개의 채널 수 정보가 담기게 됩니다.
채널이 1인 CNN 은 nn.Linear 하고 동일한 역할을 수행하여서 nn.Conv1d(a,b,1) 이렇게 하게 되면
(batch size, 채널 개수 a, input length) 가 인풋으로 들어가서 아웃풋으로 (batch size, 채널 개수 b, input length) 가 나오게 됩니다. 즉 인풋의 채널 수와 맨 마지막 output 으로 나온 채널 수를 맞춰주어서 shape 을 맞춰준 후 잔차연결을 하기 위함입니다.
self.init_weights() 의 경우,
def init_weights(self):
self.conv1.weight.data.normal_(0, 0.01)
self.conv2.weight.data.normal_(0, 0.01)
if self.downsample is not None:
self.downsample.weight.data.normal_(0, 0.01)
으로 전체 모델은 여러 residual block 으로 구성되어있고 또 각 residual block 마다 TemporalBlock 이 들어있기 때문에 residual block 개수 (num_levels 개) 만큼 for 루프를 돌면서 TemporalBlock 코드에 접근하게 됩니다. 다시 말하면 TemporalBlock 코드는 residual block 개수만큼 계속 재활용되는 것이라고 할 수 있습니다. 따라서 한번 거쳐간 후 다음 residual block 이 올 때는 그 다음 residual block 에 맞는 weight 이 새로 학습될 수 있게끔 weight 을 초기화시켜야합니다.
정리해보면 ‘CNN -> WN -> chomp ->RELU -> dropout -> CNN -> WN -> chomp -> RELU -> dropout ‘ 까지가 self.net 이고 이후 init_weights 로 weight 을 리셋해준 후 downsample 을 해야하는 경우 downsample까지 수행합니다.
def forward(self, x):
out = self.net(x)
res = x if self.downsample is None else self.downsample(x)
return self.relu(out + res)
전체 합친 TemporalBlock 코드입니다.
class TemporalBlock(nn.Module):
def __init__(self, n_inputs, n_outputs, kernel_size, stride, dilation, padding, dropout=0.2):
super(TemporalBlock, self).__init__()
self.conv1 = weight_norm(nn.Conv1d(n_inputs, n_outputs, kernel_size,
stride=stride, padding=padding, dilation=dilation))
self.chomp1 = Chomp1d(padding)
self.relu1 = nn.ReLU()
self.dropout1 = nn.Dropout(dropout)
self.conv2 = weight_norm(nn.Conv1d(n_outputs, n_outputs, kernel_size,
stride=stride, padding=padding, dilation=dilation))
self.chomp2 = Chomp1d(padding)
self.relu2 = nn.ReLU()
self.dropout2 = nn.Dropout(dropout)
self.net = nn.Sequential(self.conv1, self.chomp1, self.relu1, self.dropout1,
self.conv2, self.chomp2, self.relu2, self.dropout2)
self.downsample = nn.Conv1d(n_inputs, n_outputs, 1) if n_inputs != n_outputs else None
self.relu = nn.ReLU()
self.init_weights()
def init_weights(self):
self.conv1.weight.data.normal_(0, 0.01)
self.conv2.weight.data.normal_(0, 0.01)
if self.downsample is not None:
self.downsample.weight.data.normal_(0, 0.01)
def forward(self, x):
out = self.net(x)
res = x if self.downsample is None else self.downsample(x)
return self.relu(out + res)
TemporalConvNet
class TemporalConvNet(nn.Module):
def __init__(self, num_inputs, num_channels, kernel_size=2, dropout=0.2):
super(TemporalConvNet, self).__init__()
layers = []
num_levels = len(num_channels)
for i in range(num_levels):
dilation_size = 2 ** i
in_channels = num_inputs if i == 0 else num_channels[i-1]
out_channels = num_channels[i]
layers += [TemporalBlock(in_channels, out_channels, kernel_size, stride=1, dilation=dilation_size,
padding=(kernel_size-1) * dilation_size, dropout=dropout)]
self.network = nn.Sequential(*layers)
def forward(self, x):
return self.network(x)
nn.Sequential(*layers) 는 반복을 원하는 블록을 layers 에 추가, 추가하고 한번에 돌리는 방법이라고 합니다.
