Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/keras/src/layers/convolutional/base_conv.py: 18%
140 statements
« prev ^ index » next coverage.py v7.4.0, created at 2024-01-03 07:57 +0000
1# Copyright 2015 The TensorFlow Authors. All Rights Reserved.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14# ==============================================================================
15"""Keras base class for convolution layers."""
18import tensorflow.compat.v2 as tf
20from keras.src import activations
21from keras.src import constraints
22from keras.src import initializers
23from keras.src import regularizers
24from keras.src.engine.base_layer import Layer
25from keras.src.engine.input_spec import InputSpec
26from keras.src.utils import conv_utils
29class Conv(Layer):
30 """Abstract N-D convolution layer (private, used as implementation base).
32 This layer creates a convolution kernel that is convolved
33 (actually cross-correlated) with the layer input to produce a tensor of
34 outputs. If `use_bias` is True (and a `bias_initializer` is provided),
35 a bias vector is created and added to the outputs. Finally, if
36 `activation` is not `None`, it is applied to the outputs as well.
38 Note: layer attributes cannot be modified after the layer has been called
39 once (except the `trainable` attribute).
41 Args:
42 rank: An integer, the rank of the convolution, e.g. "2" for 2D
43 convolution.
44 filters: Integer, the dimensionality of the output space (i.e. the number
45 of filters in the convolution). Could be "None", eg in the case of
46 depth wise convolution.
47 kernel_size: An integer or tuple/list of n integers, specifying the
48 length of the convolution window.
49 strides: An integer or tuple/list of n integers,
50 specifying the stride length of the convolution.
51 Specifying any stride value != 1 is incompatible with specifying
52 any `dilation_rate` value != 1.
53 padding: One of `"valid"`, `"same"`, or `"causal"` (case-insensitive).
54 `"valid"` means no padding. `"same"` results in padding with zeros
55 evenly to the left/right or up/down of the input such that output has
56 the same height/width dimension as the input. `"causal"` results in
57 causal (dilated) convolutions, e.g. `output[t]` does not depend on
58 `input[t+1:]`.
59 data_format: A string, one of `channels_last` (default) or
60 `channels_first`.
61 The ordering of the dimensions in the inputs.
62 `channels_last` corresponds to inputs with shape
63 `(batch_size, ..., channels)` while `channels_first` corresponds to
64 inputs with shape `(batch_size, channels, ...)`.
65 dilation_rate: An integer or tuple/list of n integers, specifying
66 the dilation rate to use for dilated convolution.
67 Currently, specifying any `dilation_rate` value != 1 is
68 incompatible with specifying any `strides` value != 1.
69 groups: A positive integer specifying the number of groups in which the
70 input is split along the channel axis. Each group is convolved
71 separately with `filters / groups` filters. The output is the
72 concatenation of all the `groups` results along the channel axis.
73 Input channels and `filters` must both be divisible by `groups`.
74 activation: Activation function to use.
75 If you don't specify anything, no activation is applied.
76 use_bias: Boolean, whether the layer uses a bias.
77 kernel_initializer: An initializer for the convolution kernel. If None,
78 the default initializer (glorot_uniform) will be used.
79 bias_initializer: An initializer for the bias vector. If None, the default
80 initializer (zeros) will be used.
81 kernel_regularizer: Optional regularizer for the convolution kernel.
82 bias_regularizer: Optional regularizer for the bias vector.
83 activity_regularizer: Optional regularizer function for the output.
84 kernel_constraint: Optional projection function to be applied to the
85 kernel after being updated by an `Optimizer` (e.g. used to implement
86 norm constraints or value constraints for layer weights). The function
87 must take as input the unprojected variable and must return the
88 projected variable (which must have the same shape). Constraints are
89 not safe to use when doing asynchronous distributed training.
90 bias_constraint: Optional projection function to be applied to the
91 bias after being updated by an `Optimizer`.
92 """
94 def __init__(
95 self,
96 rank,
97 filters,
98 kernel_size,
99 strides=1,
100 padding="valid",
101 data_format=None,
102 dilation_rate=1,
103 groups=1,
104 activation=None,
105 use_bias=True,
106 kernel_initializer="glorot_uniform",
107 bias_initializer="zeros",
108 kernel_regularizer=None,
109 bias_regularizer=None,
110 activity_regularizer=None,
111 kernel_constraint=None,
112 bias_constraint=None,
113 trainable=True,
114 name=None,
115 conv_op=None,
116 **kwargs,
117 ):
118 super().__init__(
119 trainable=trainable,
120 name=name,
121 activity_regularizer=regularizers.get(activity_regularizer),
122 **kwargs,
123 )
124 self.rank = rank
126 if isinstance(filters, float):
127 filters = int(filters)
128 if filters is not None and filters <= 0:
129 raise ValueError(
130 "Invalid value for argument `filters`. "
131 "Expected a strictly positive value. "
132 f"Received filters={filters}."
133 )
134 self.filters = filters
135 self.groups = groups or 1
136 self.kernel_size = conv_utils.normalize_tuple(
137 kernel_size, rank, "kernel_size"
138 )
139 self.strides = conv_utils.normalize_tuple(
140 strides, rank, "strides", allow_zero=True
141 )
142 self.padding = conv_utils.normalize_padding(padding)
143 self.data_format = conv_utils.normalize_data_format(data_format)
144 self.dilation_rate = conv_utils.normalize_tuple(
145 dilation_rate, rank, "dilation_rate"
146 )
148 self.activation = activations.get(activation)
149 self.use_bias = use_bias
151 self.kernel_initializer = initializers.get(kernel_initializer)
152 self.bias_initializer = initializers.get(bias_initializer)
153 self.kernel_regularizer = regularizers.get(kernel_regularizer)
154 self.bias_regularizer = regularizers.get(bias_regularizer)
155 self.kernel_constraint = constraints.get(kernel_constraint)
156 self.bias_constraint = constraints.get(bias_constraint)
157 self.input_spec = InputSpec(min_ndim=self.rank + 2)
159 self._validate_init()
160 self._is_causal = self.padding == "causal"
161 self._channels_first = self.data_format == "channels_first"
162 self._tf_data_format = conv_utils.convert_data_format(
163 self.data_format, self.rank + 2
164 )
166 def _validate_init(self):
167 if self.filters is not None and self.filters % self.groups != 0:
168 raise ValueError(
169 "The number of filters must be evenly divisible by the "
170 "number of groups. Received: groups={}, filters={}".format(
171 self.groups, self.filters
172 )
173 )
175 if not all(self.kernel_size):
176 raise ValueError(
177 "The argument `kernel_size` cannot contain 0(s). Received: %s"
178 % (self.kernel_size,)
179 )
181 if not all(self.strides):
182 raise ValueError(
183 "The argument `strides` cannot contains 0(s). Received: %s"
184 % (self.strides,)
185 )
187 if self.padding == "causal":
189 from keras.src.layers.convolutional.conv1d import Conv1D
190 from keras.src.layers.convolutional.separable_conv1d import (
191 SeparableConv1D,
192 )
194 if not isinstance(self, (Conv1D, SeparableConv1D)):
195 raise ValueError(
196 "Causal padding is only supported for `Conv1D`"
197 "and `SeparableConv1D`."
198 )
200 if max(self.strides) > 1 and max(self.dilation_rate) > 1:
201 raise ValueError(
202 "`strides > 1` not supported in conjunction with "
203 f"`dilation_rate > 1`. Received: strides={self.strides} and "
204 f"dilation_rate={self.dilation_rate}"
205 )
207 def build(self, input_shape):
208 input_shape = tf.TensorShape(input_shape)
209 input_channel = self._get_input_channel(input_shape)
210 if input_channel % self.groups != 0:
211 raise ValueError(
212 "The number of input channels must be evenly divisible by "
213 "the number of groups. Received groups={}, but the input "
214 "has {} channels (full input shape is {}).".format(
215 self.groups, input_channel, input_shape
216 )
217 )
218 kernel_shape = self.kernel_size + (
219 input_channel // self.groups,
220 self.filters,
221 )
223 # compute_output_shape contains some validation logic for the input
224 # shape, and make sure the output shape has all positive dimensions.
225 self.compute_output_shape(input_shape)
227 self.kernel = self.add_weight(
228 name="kernel",
229 shape=kernel_shape,
230 initializer=self.kernel_initializer,
231 regularizer=self.kernel_regularizer,
232 constraint=self.kernel_constraint,
233 trainable=True,
234 dtype=self.dtype,
235 )
236 if self.use_bias:
237 self.bias = self.add_weight(
238 name="bias",
239 shape=(self.filters,),
240 initializer=self.bias_initializer,
241 regularizer=self.bias_regularizer,
242 constraint=self.bias_constraint,
243 trainable=True,
244 dtype=self.dtype,
245 )
246 else:
247 self.bias = None
248 channel_axis = self._get_channel_axis()
249 self.input_spec = InputSpec(
250 min_ndim=self.rank + 2, axes={channel_axis: input_channel}
251 )
252 self.built = True
254 def convolution_op(self, inputs, kernel):
255 if self.padding == "causal":
256 tf_padding = "VALID" # Causal padding handled in `call`.
257 elif isinstance(self.padding, str):
258 tf_padding = self.padding.upper()
259 else:
260 tf_padding = self.padding
262 return tf.nn.convolution(
263 inputs,
264 kernel,
265 strides=list(self.strides),
266 padding=tf_padding,
267 dilations=list(self.dilation_rate),
268 data_format=self._tf_data_format,
269 name=self.__class__.__name__,
270 )
272 # TODO(b/213173659): remove this when grouped convolutions are fully
273 # supported on the CPU for compiled functions. For now, we need this as a
274 # workaround for CPU support.
275 @tf.function(jit_compile=True)
276 def _jit_compiled_convolution_op(self, inputs, kernel):
277 return self.convolution_op(inputs, kernel)
279 def call(self, inputs):
280 input_shape = inputs.shape
282 if self._is_causal: # Apply causal padding to inputs for Conv1D.
283 inputs = tf.pad(inputs, self._compute_causal_padding(inputs))
285 if self.groups > 1:
286 outputs = self._jit_compiled_convolution_op(
287 inputs, tf.convert_to_tensor(self.kernel)
288 )
289 else:
290 outputs = self.convolution_op(inputs, self.kernel)
292 if self.use_bias:
293 output_rank = outputs.shape.rank
294 if self.rank == 1 and self._channels_first:
295 # nn.bias_add does not accept a 1D input tensor.
296 bias = tf.reshape(self.bias, (1, self.filters, 1))
297 outputs += bias
298 else:
299 # Handle multiple batch dimensions.
300 if output_rank is not None and output_rank > 2 + self.rank:
302 def _apply_fn(o):
303 return tf.nn.bias_add(
304 o, self.bias, data_format=self._tf_data_format
305 )
307 outputs = conv_utils.squeeze_batch_dims(
308 outputs, _apply_fn, inner_rank=self.rank + 1
309 )
310 else:
311 outputs = tf.nn.bias_add(
312 outputs, self.bias, data_format=self._tf_data_format
313 )
315 if not tf.executing_eagerly() and input_shape.rank:
316 # Infer the static output shape:
317 out_shape = self.compute_output_shape(input_shape)
318 outputs.set_shape(out_shape)
320 if self.activation is not None:
321 return self.activation(outputs)
322 return outputs
324 def _spatial_output_shape(self, spatial_input_shape):
325 return [
326 conv_utils.conv_output_length(
327 length,
328 self.kernel_size[i],
329 padding=self.padding,
330 stride=self.strides[i],
331 dilation=self.dilation_rate[i],
332 )
333 for i, length in enumerate(spatial_input_shape)
334 ]
336 def compute_output_shape(self, input_shape):
337 input_shape = tf.TensorShape(input_shape).as_list()
338 batch_rank = len(input_shape) - self.rank - 1
339 try:
340 if self.data_format == "channels_last":
341 return tf.TensorShape(
342 input_shape[:batch_rank]
343 + self._spatial_output_shape(input_shape[batch_rank:-1])
344 + [self.filters]
345 )
346 else:
347 return tf.TensorShape(
348 input_shape[:batch_rank]
349 + [self.filters]
350 + self._spatial_output_shape(input_shape[batch_rank + 1 :])
351 )
353 except ValueError:
354 raise ValueError(
355 "One of the dimensions in the output is <= 0 "
356 f"due to downsampling in {self.name}. Consider "
357 "increasing the input size. "
358 f"Received input shape {input_shape} which would produce "
359 "output shape with a zero or negative value in a "
360 "dimension."
361 )
363 def _recreate_conv_op(self, inputs):
364 return False
366 def get_config(self):
367 config = {
368 "filters": self.filters,
369 "kernel_size": self.kernel_size,
370 "strides": self.strides,
371 "padding": self.padding,
372 "data_format": self.data_format,
373 "dilation_rate": self.dilation_rate,
374 "groups": self.groups,
375 "activation": activations.serialize(self.activation),
376 "use_bias": self.use_bias,
377 "kernel_initializer": initializers.serialize(
378 self.kernel_initializer
379 ),
380 "bias_initializer": initializers.serialize(self.bias_initializer),
381 "kernel_regularizer": regularizers.serialize(
382 self.kernel_regularizer
383 ),
384 "bias_regularizer": regularizers.serialize(self.bias_regularizer),
385 "activity_regularizer": regularizers.serialize(
386 self.activity_regularizer
387 ),
388 "kernel_constraint": constraints.serialize(self.kernel_constraint),
389 "bias_constraint": constraints.serialize(self.bias_constraint),
390 }
391 base_config = super().get_config()
392 return dict(list(base_config.items()) + list(config.items()))
394 def _compute_causal_padding(self, inputs):
395 """Calculates padding for 'causal' option for 1-d conv layers."""
396 left_pad = self.dilation_rate[0] * (self.kernel_size[0] - 1)
397 if getattr(inputs.shape, "ndims", None) is None:
398 batch_rank = 1
399 else:
400 batch_rank = len(inputs.shape) - 2
401 if self.data_format == "channels_last":
402 causal_padding = [[0, 0]] * batch_rank + [[left_pad, 0], [0, 0]]
403 else:
404 causal_padding = [[0, 0]] * batch_rank + [[0, 0], [left_pad, 0]]
405 return causal_padding
407 def _get_channel_axis(self):
408 if self.data_format == "channels_first":
409 return -1 - self.rank
410 else:
411 return -1
413 def _get_input_channel(self, input_shape):
414 channel_axis = self._get_channel_axis()
415 if input_shape.dims[channel_axis].value is None:
416 raise ValueError(
417 "The channel dimension of the inputs should be defined. "
418 f"The input_shape received is {input_shape}, "
419 f"where axis {channel_axis} (0-based) "
420 "is the channel dimension, which found to be `None`."
421 )
422 return int(input_shape[channel_axis])
424 def _get_padding_op(self):
425 if self.padding == "causal":
426 op_padding = "valid"
427 else:
428 op_padding = self.padding
429 if not isinstance(op_padding, (list, tuple)):
430 op_padding = op_padding.upper()
431 return op_padding