既存レイヤーをまとめたラッパーレイヤーが作りたい
繰り返し構造を用いたネットワークを作成する際に、繰り返しブロックを関数化して、コールしているソースはよくみるが、
せっかくFunctional API1で書いているのに、途中で内部関数呼んでて何やら見辛い。
deep_unet.py
class DeepUNet():
def __init__(self, input_shape, internal_filter=32, depth=7, class_num=3):
inputs = Input(self.__input_shape)
conv_layer = Conv2D(internal_filter, 3, padding='same')(inputs)
puls_layers = []
encode_layer = conv_layer
for d in range(depth):
block_outputs = self.__down_block(encode_layer, internal_filter=internal_filter*2)
encode_layer = block_outputs[0]
puls_layers.append(block_outputs[1])
decode_layer = encode_layer
for puls_layer in reversed(puls_layers):
decode_layer = self.__up_block([decode_layer, puls_layer], internal_filter=internal_filter*2)
decode_layer = BatchNormalization()(decode_layer)
decode_layer = Activation('relu')(decode_layer)
outputs = Conv2D(class_num, 1, activation='softmax', padding='same')(decode_layer)
self.__model = Model(inputs=[inputs], outputs=[outputs])
def __down_block(self, input_layer, internal_filter=64):
filters = input_layer.get_shape().as_list()[3]
conv_1 = Conv2D(internal_filter, kernel_size, strides=strides, padding='same')(input_layer)
conv_2 = Conv2D(filters, kernel_size, strides=strides, padding='same')(conv_1)
puls = Add()([conv_2, input_layer])
pool = MaxPooling2D()(puls)
return pool, puls
def __up_block(self, input_layers, concat_layer, internal_filter=64):
filters = input_layer.get_shape().as_list()[3]
upsample = UpSampling2D()(input_layer)
act = Activation('relu')(upsample)
concat_layer = Concatenate()([act, concat_layer])
conv_1 = Conv2D(self.__internal_filter, 3, strides=1, padding='same')(concat_layer)
conv_2 = Conv2D(filters, 3, strides=1, padding='same')(conv_1)
puls = Add()([conv_2, upsample])
return puls
よくみるのは、こんな↑感じ。
block_outputs = self.__down_block(~~)とか、decode_layer = self.__up_block(~~)が非常に微妙。(に感じる2)
deep_unet.py
class DeepUNet():
def __init__(self, input_shape, internal_filter=32, depth=7, class_num=3):
self.__input_shape = input_shape
self.__class_num = class_num
inputs = Input(self.__input_shape)
conv_layer = Conv2D(internal_filter, 3, padding='same')(inputs)
puls_layers = []
encode_layer = conv_layer
for d in range(depth):
block_outputs = DownBlock(internal_filter=internal_filter*2)(encode_layer)
encode_layer = block_outputs[0]
puls_layers.append(block_outputs[1])
decode_layer = encode_layer
for puls_layer in reversed(puls_layers):
decode_layer = UpBlock(internal_filter=internal_filter*2)([decode_layer, puls_layer])
decode_layer = BatchNormalization()(decode_layer)
decode_layer = Activation('relu')(decode_layer)
outputs = Conv2D(class_num, 1, activation='softmax', padding='same')(decode_layer)
self.__model = Model(inputs=[inputs], outputs=[outputs])
~~ 省略 ~~
各BlockをLayer化 (失敗)
まずは、普通に、各Blockをkeras.engine.base_layerのLayerを継承させて、カスタムレイヤー化することを考えて実装してみた。
layers/down_block.py
import keras.engine.base_layer as KELayer
~~ 省略 ~~
class DownBlock(KELayer.Layer):
def __init__(self, internal_filter=64, **kwargs):
super(DownBlock, self).__init__(**kwargs)
self.__internal_filter = internal_filter
def build(self, input_shape):
super(DownBlock, self).build(input_shape)
def call(self, inputs, **kwargs):
return self.__down_block(inputs)
def __down_block(self, input_layer):
filters = input_layer.get_shape().as_list()[3]
conv_1 = Conv2D(self.__internal_filter, kernel_size, strides=strides, padding='same')(input_layer)
conv_2 = Conv2D(filters, kernel_size, strides=strides, padding='same')(conv_1)
puls = Add()([conv_2, input_layer])
pool = MaxPooling2D()(puls)
return [pool, puls]
def compute_output_shape(self, input_shape):
return [ (None, input_shape[1] // 2, input_shape[2] // 2, input_shape[3])
, (None, input_shape[1], input_shape[2], input_shape[3])
]
layers/up_block.py
import keras.engine.base_layer as KELayer
~~ 省略 ~~
class UpBlock(KELayer.Layer):
def __init__(self, internal_filter=64, **kwargs):
super(UpBlock, self).__init__(**kwargs)
self.__internal_filter = internal_filter
def build(self, input_shape):
super(UpBlock, self).build(input_shape)
def call(self, inputs, **kwargs):
return self.__up_block(*inputs)
def __up_block(self, input_layer, concat_layer):
filters = input_layer.get_shape().as_list()[3]
upsample = UpSampling2D()(input_layer)
act = Activation('relu')(upsample)
concat_layer = Concatenate()([act, concat_layer])
conv_1 = Conv2D(self.__internal_filter, 3, strides=1, padding='same')(concat_layer)
conv_2 = Conv2D(filters, 3, strides=1, padding='same')(conv_1)
puls = Add()([conv_2, upsample])
return puls
def compute_output_shape(self, input_shape):
return [(None, input_shape[0][1] * 2, input_shape[0][2] * 2, input_shape[0][3])]
と実装して、いざ学習してみると、結果が思わしくない。。。
model.summaryしてみると (depth=5です)
__________________________________________________________________________________________________
Layer (type) Output Shape Param # Connected to
==================================================================================================
input_1 (InputLayer) (None, 256, 256, 3) 0
__________________________________________________________________________________________________
conv2d_1 (Conv2D) (None, 256, 256, 64) 1792 input_1[0][0]
__________________________________________________________________________________________________
down_block_1 (DownBlock) [(None, 128, 128, 64 0 conv2d_1[0][0]
__________________________________________________________________________________________________
down_block_2 (DownBlock) [(None, 64, 64, 64), 0 down_block_1[0][0]
__________________________________________________________________________________________________
down_block_3 (DownBlock) [(None, 32, 32, 64), 0 down_block_2[0][0]
__________________________________________________________________________________________________
down_block_4 (DownBlock) [(None, 16, 16, 64), 0 down_block_3[0][0]
__________________________________________________________________________________________________
down_block_5 (DownBlock) [(None, 8, 8, 64), ( 0 down_block_4[0][0]
__________________________________________________________________________________________________
up_block_1 (UpBlock) (None, 16, 16, 64) 0 down_block_5[0][0]
down_block_5[0][1]
__________________________________________________________________________________________________
up_block_2 (UpBlock) (None, 32, 32, 64) 0 up_block_1[0][0]
down_block_4[0][1]
__________________________________________________________________________________________________
up_block_3 (UpBlock) (None, 64, 64, 64) 0 up_block_2[0][0]
down_block_3[0][1]
__________________________________________________________________________________________________
up_block_4 (UpBlock) (None, 128, 128, 64) 0 up_block_3[0][0]
down_block_2[0][1]
__________________________________________________________________________________________________
up_block_5 (UpBlock) (None, 256, 256, 64) 0 up_block_4[0][0]
down_block_1[0][1]
__________________________________________________________________________________________________
batch_normalization_21 (BatchNo (None, 256, 256, 64) 256 up_block_5[0][0]
__________________________________________________________________________________________________
activation_21 (Activation) (None, 256, 256, 64) 0 batch_normalization_21[0][0]
__________________________________________________________________________________________________
conv2d_22 (Conv2D) (None, 256, 256, 3) 195 activation_21[0][0]
==================================================================================================
Total params: 2,243
Trainable params: 2,115
Non-trainable params: 128
__________________________________________________________________________________________________
あぁ、そうか、add_weightしてないからか。。。
で、調べてみたが、自分で作った処理のadd_weightの話は出てくるが、
内部で既存レイヤー使ったときの話が出て来ず、、、
(そんな需要がない or カスタムレイヤーってそういうんじゃねぇからってことですかね)
各Blockを擬似Layer化(CallableなClass化) (とりあえず成功)
ということで、困っていたが、「あぁそうか、（自分のしたいことの為には）別にLayerの継承にこだわらなくていいか」ってことで、
各Blockを、Layerを継承せずCallableなClassにすることで、擬似Layer化することに。
layers/down_block.py
~~ 省略 ~~
class DownBlock():
def __init__(self, internal_filter=64):
self.__internal_filter = internal_filter
def __call__(self, inputs):
return self.__down_block(inputs)
def __down_block(self, input_layer):
filters = input_layer.get_shape().as_list()[3]
conv_1 = Conv2D(self.__internal_filter, kernel_size, strides=strides, padding='same')(input_layer)
conv_2 = Conv2D(filters, kernel_size, strides=strides, padding='same')(conv_1)
puls = Add()([conv_2, input_layer])
pool = MaxPooling2D()(puls)
return [pool, puls]
layers/up_block.py
~~ 省略 ~~
class UpBlock():
def __init__(self, internal_filter=64):
self.__internal_filter = internal_filter
def __call__(self, inputs):
return self.__up_block(*inputs)
def __up_block(self, input_layer, concat_layer):
filters = input_layer.get_shape().as_list()[3]
upsample = UpSampling2D()(input_layer)
act = Activation('relu')(upsample)
concat_layer = Concatenate()([act, concat_layer])
conv_1 = Conv2D(self.__internal_filter, 3, strides=1, padding='same')(concat_layer)
conv_2 = Conv2D(filters, 3, strides=1, padding='same')(conv_1)
puls = Add()([conv_2, upsample])
return puls
これで、学習もうまく行きました。
ただし、こちらの場合、擬似Layerなので、model.Summaryとかしても、（この擬似Layerは）表示されず、
中身がダダァーッと表示されます。(これの方がいいこともありますが。(内部がわかるんで。))
グラフ化するときも注意です。
最後に
Layerを継承する形でもうまく行く方法を知っている方いましたら、教えていただけませんか？＞＜