Coverage for mlair/model_modules/inception_model.py: 100%
82 statements
« prev ^ index » next coverage.py v6.4.2, created at 2023-06-30 10:22 +0000
1__author__ = 'Felix Kleinert, Lukas Leufen'
2__date__ = '2019-10-22'
4import logging
6import tensorflow.keras as keras
7import tensorflow.keras.layers as layers
9from mlair.model_modules.advanced_paddings import PadUtils, ReflectionPadding2D, Padding2D
12class InceptionModelBase:
13 """
14 This class contains all necessary construction blocks
15 """
17 def __init__(self):
18 self.number_of_blocks = 0
19 self.part_of_block = 0
20 self.act_number = 0
21 self.ord_base = 96 # set to 96 as always add +1 for new part of block, chr(97)='a'
23 def block_part_name(self):
24 """
25 Use unicode due to some issues of keras with normal strings
27 :return:
28 """
29 return chr(self.ord_base + self.part_of_block)
31 def batch_normalisation(self, input_x, **kwargs):
32 block_name = f"Block_{self.number_of_blocks}{self.block_part_name()}_BN"
33 return layers.BatchNormalization(name=block_name, **kwargs)(input_x)
35 def create_conv_tower(self,
36 input_x,
37 reduction_filter,
38 tower_filter,
39 tower_kernel,
40 activation='relu',
41 batch_normalisation=False,
42 **kwargs):
43 """
44 This function creates a "convolution tower block" containing a 1x1 convolution to reduce filter size followed by
45 convolution with given filter and kernel size
47 :param input_x: Input to network part
48 :param reduction_filter: Number of filters used in 1x1 convolution to reduce overall filter size before conv.
49 :param tower_filter: Number of filters for n x m convolution
50 :param tower_kernel: kernel size for convolution (n,m)
51 :param activation: activation function for convolution
52 :param batch_normalisation:
53 :return:
54 """
55 self.part_of_block += 1
56 self.act_number = 1
57 regularizer = kwargs.get('regularizer', keras.regularizers.l2(0.01))
58 bn_settings = kwargs.get('bn_settings', {})
59 act_settings = kwargs.get('act_settings', {})
60 padding = kwargs.get('padding', 'ZeroPad2D')
61 logging.debug(f'Inception Block with activation: {activation}')
63 block_name = f'Block_{self.number_of_blocks}{self.block_part_name()}_{tower_kernel[0]}x{tower_kernel[1]}'
64 padding_size = PadUtils.get_padding_for_same(tower_kernel)
66 if tower_kernel == (1, 1):
67 tower = layers.Conv2D(tower_filter,
68 tower_kernel,
69 padding='valid',
70 kernel_regularizer=regularizer,
71 name=block_name)(input_x)
72 # tower = self.act(tower, activation, **act_settings)
73 else:
74 tower = layers.Conv2D(reduction_filter,
75 (1, 1),
76 padding='valid',
77 kernel_regularizer=regularizer,
78 name=f'Block_{self.number_of_blocks}{self.block_part_name()}_1x1')(input_x)
79 tower = self.act(tower, activation, **act_settings)
81 tower = Padding2D(padding)(padding=padding_size,
82 name=f'Block_{self.number_of_blocks}{self.block_part_name()}_Pad'
83 )(tower)
85 tower = layers.Conv2D(tower_filter,
86 tower_kernel,
87 padding='valid',
88 kernel_regularizer=regularizer,
89 name=block_name)(tower)
90 if batch_normalisation:
91 tower = self.batch_normalisation(tower, **bn_settings)
92 tower = self.act(tower, activation, **act_settings)
94 return tower
96 def act(self, input_x, activation, **act_settings):
97 block_name = f"Block_{self.number_of_blocks}{self.block_part_name()}_act_{self.act_number}"
98 try:
99 out = getattr(layers, self._get_act_name(activation))(**act_settings, name=block_name)(input_x)
100 except AttributeError:
101 block_name += f"_{activation.lower()}"
102 out = layers.Activation(activation.lower(), name=block_name)(input_x)
103 self.act_number += 1
104 return out
106 @staticmethod
107 def _get_act_name(act_name):
108 if isinstance(act_name, str):
109 mapping = {'relu': 'ReLU', 'prelu': 'PReLU', 'elu': 'ELU'}
110 return mapping.get(act_name.lower(), act_name)
111 else:
112 return act_name.__name__
114 def create_pool_tower(self, input_x, pool_kernel, tower_filter, activation='relu', max_pooling=True, **kwargs):
115 """
116 This function creates a "MaxPooling tower block"
118 :param input_x: Input to network part
119 :param pool_kernel: size of pooling kernel
120 :param tower_filter: Number of filters used in 1x1 convolution to reduce filter size
121 :param activation:
122 :param max_pooling:
123 :return:
124 """
125 self.part_of_block += 1
126 self.act_number = 1
127 act_settings = kwargs.get('act_settings', {})
128 padding = kwargs.get('padding', 'ZeroPad2D')
129 padding_size = PadUtils.get_padding_for_same(kernel_size=pool_kernel)
131 # pooling block
132 block_name = f"Block_{self.number_of_blocks}{self.block_part_name()}_"
133 if max_pooling:
134 block_type = "MaxPool"
135 pooling = layers.MaxPooling2D
136 else:
137 block_type = "AvgPool"
138 pooling = layers.AveragePooling2D
140 tower = Padding2D(padding)(padding=padding_size, name=block_name + 'Pad')(input_x)
141 tower = pooling(pool_kernel, strides=(1, 1), padding='valid', name=block_name + block_type)(tower)
143 # convolution block
144 tower = layers.Conv2D(tower_filter, (1, 1), padding='valid', name=block_name + "1x1")(tower)
145 tower = self.act(tower, activation, **act_settings)
147 return tower
149 def inception_block(self, input_x, tower_conv_parts, tower_pool_parts, **kwargs):
150 """
151 Crate a inception block
153 :param input_x: Input to block
154 :param tower_conv_parts: dict containing settings for parts of inception block; Example:
155 tower_conv_parts = {'tower_1': {'reduction_filter': 32,
156 'tower_filter': 64,
157 'tower_kernel': (3, 1),
158 'activation' : 'relu',
159 'padding' : 'SymPad2D'}
160 'tower_2': {'reduction_filter': 32,
161 'tower_filter': 64,
162 'tower_kernel': (5, 1),
163 'activation' : LeakyReLU,
164 'padding' : keras.layers.ZeroPadding2D}
165 'tower_3': {'reduction_filter': 32,
166 'tower_filter': 64,
167 'tower_kernel': (1, 1),
168 'activation' : ELU,
169 'padding' : src.model_modules.advanced_paddings.ReflectionPadding2D}
170 }
171 :param tower_pool_parts: dict containing settings for pool part of inception block; Example:
172 tower_pool_parts = {'pool_kernel': (3, 1), 'tower_filter': 64, 'padding': 'RefPad2D'}
173 :return:
174 """
175 self.number_of_blocks += 1
176 self.part_of_block = 0
177 tower_build = {}
178 block_name = f"Block_{self.number_of_blocks}"
179 for part, part_settings in tower_conv_parts.items():
180 tower_build[part] = self.create_conv_tower(input_x, **part_settings, **kwargs)
181 if 'max_pooling' in tower_pool_parts.keys():
182 max_pooling = tower_pool_parts.get('max_pooling')
183 if not isinstance(max_pooling, bool):
184 raise AttributeError(f"max_pooling has to be either a bool or empty. Given was: {max_pooling}")
185 pool_name = '{}pool'.format('max' if max_pooling else 'avg')
186 tower_build[pool_name] = self.create_pool_tower(input_x, **tower_pool_parts, **kwargs)
187 else:
188 tower_build['maxpool'] = self.create_pool_tower(input_x, **tower_pool_parts, **kwargs)
189 tower_build['avgpool'] = self.create_pool_tower(input_x, **tower_pool_parts, **kwargs, max_pooling=False)
191 block = keras.layers.concatenate(list(tower_build.values()), axis=3,
192 name=block_name + "_Co")
193 return block
196if __name__ == '__main__':
197 print(__name__)
198 from keras.datasets import cifar10
199 from keras.utils import np_utils
200 from keras.layers import Input
201 from keras.layers.advanced_activations import LeakyReLU, ELU
202 from keras.optimizers import SGD
203 from keras.layers import Dense, Flatten, Conv2D, MaxPooling2D
204 from keras.models import Model
206 # network settings
207 conv_settings_dict = {'tower_1': {'reduction_filter': 64,
208 'tower_filter': 64,
209 'tower_kernel': (3, 3),
210 'activation': LeakyReLU, },
211 'tower_2': {'reduction_filter': 64,
212 'tower_filter': 64,
213 'tower_kernel': (5, 5),
214 'activation': 'relu',
215 'padding': 'SymPad2D'},
216 'tower_3': {'reduction_filter': 64,
217 'tower_filter': 64,
218 'tower_kernel': (7, 7),
219 'activation': ELU,
220 'padding': ReflectionPadding2D}
221 }
222 pool_settings_dict = {'pool_kernel': (3, 3),
223 'tower_filter': 64,
224 'activation': 'relu'}
226 # load data
227 (X_train, y_train), (X_test, y_test) = cifar10.load_data()
228 X_train = X_train.astype('float32')
229 X_test = X_test.astype('float32')
230 X_train = X_train / 255.0
231 X_test = X_test / 255.0
232 y_train = np_utils.to_categorical(y_train)
233 y_test = np_utils.to_categorical(y_test)
234 input_img = Input(shape=(32, 32, 3))
236 # create inception net
237 inception_net = InceptionModelBase()
238 output = inception_net.inception_block(input_img, conv_settings_dict, pool_settings_dict, batch_normalisation=True)
239 output = Flatten()(output)
240 output = Dense(10, activation='softmax')(output)
241 model = Model(inputs=input_img, outputs=output)
242 print(model.summary())
244 # compile
245 epochs = 1
246 lrate = 0.01
247 decay = lrate / epochs
248 sgd = SGD(lr=lrate, momentum=0.9, decay=decay, nesterov=False)
249 model.compile(loss='categorical_crossentropy', optimizer=sgd, metrics=['accuracy'])
250 print(X_train.shape)
251 keras.utils.plot_model(model, to_file='model.pdf', show_shapes=True, show_layer_names=True)
252 # model.fit(X_train, y_train, epochs=epochs, validation_data=(X_test, y_test))
253 print('test')