Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/keras/src/layers/preprocessing/normalization.py: 19%
112 statements
« prev ^ index » next coverage.py v7.4.0, created at 2024-01-03 07:57 +0000
1# Copyright 2019 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"""Normalization preprocessing layer."""
18import numpy as np
19import tensorflow.compat.v2 as tf
21from keras.src import backend
22from keras.src.engine import base_preprocessing_layer
23from keras.src.layers.preprocessing import preprocessing_utils as utils
25# isort: off
26from tensorflow.python.util.tf_export import keras_export
29@keras_export(
30 "keras.layers.Normalization",
31 "keras.layers.experimental.preprocessing.Normalization",
32)
33class Normalization(base_preprocessing_layer.PreprocessingLayer):
34 """A preprocessing layer which normalizes continuous features.
36 This layer will shift and scale inputs into a distribution centered around
37 0 with standard deviation 1. It accomplishes this by precomputing the mean
38 and variance of the data, and calling `(input - mean) / sqrt(var)` at
39 runtime.
41 The mean and variance values for the layer must be either supplied on
42 construction or learned via `adapt()`. `adapt()` will compute the mean and
43 variance of the data and store them as the layer's weights. `adapt()` should
44 be called before `fit()`, `evaluate()`, or `predict()`.
46 For an overview and full list of preprocessing layers, see the preprocessing
47 [guide](https://www.tensorflow.org/guide/keras/preprocessing_layers).
49 Args:
50 axis: Integer, tuple of integers, or None. The axis or axes that should
51 have a separate mean and variance for each index in the shape. For
52 example, if shape is `(None, 5)` and `axis=1`, the layer will track 5
53 separate mean and variance values for the last axis. If `axis` is set
54 to `None`, the layer will normalize all elements in the input by a
55 scalar mean and variance. When `-1` the last axis of the
56 input is assumed to be a feature dimension and is normalized per
57 index. Note that in the specific case of batched scalar inputs where
58 the only axis is the batch axis, the default will normalize each index
59 in the batch separately. In this case, consider passing `axis=None`.
60 Defaults to `-1`.
61 mean: The mean value(s) to use during normalization. The passed value(s)
62 will be broadcast to the shape of the kept axes above; if the value(s)
63 cannot be broadcast, an error will be raised when this layer's
64 `build()` method is called.
65 variance: The variance value(s) to use during normalization. The passed
66 value(s) will be broadcast to the shape of the kept axes above; if the
67 value(s) cannot be broadcast, an error will be raised when this
68 layer's `build()` method is called.
69 invert: If True, this layer will apply the inverse transformation
70 to its inputs: it would turn a normalized input back into its
71 original form.
73 Examples:
75 Calculate a global mean and variance by analyzing the dataset in `adapt()`.
77 >>> adapt_data = np.array([1., 2., 3., 4., 5.], dtype='float32')
78 >>> input_data = np.array([1., 2., 3.], dtype='float32')
79 >>> layer = tf.keras.layers.Normalization(axis=None)
80 >>> layer.adapt(adapt_data)
81 >>> layer(input_data)
82 <tf.Tensor: shape=(3,), dtype=float32, numpy=
83 array([-1.4142135, -0.70710677, 0.], dtype=float32)>
85 Calculate a mean and variance for each index on the last axis.
87 >>> adapt_data = np.array([[0., 7., 4.],
88 ... [2., 9., 6.],
89 ... [0., 7., 4.],
90 ... [2., 9., 6.]], dtype='float32')
91 >>> input_data = np.array([[0., 7., 4.]], dtype='float32')
92 >>> layer = tf.keras.layers.Normalization(axis=-1)
93 >>> layer.adapt(adapt_data)
94 >>> layer(input_data)
95 <tf.Tensor: shape=(1, 3), dtype=float32, numpy=
96 array([-1., -1., -1.], dtype=float32)>
98 Pass the mean and variance directly.
100 >>> input_data = np.array([[1.], [2.], [3.]], dtype='float32')
101 >>> layer = tf.keras.layers.Normalization(mean=3., variance=2.)
102 >>> layer(input_data)
103 <tf.Tensor: shape=(3, 1), dtype=float32, numpy=
104 array([[-1.4142135 ],
105 [-0.70710677],
106 [ 0. ]], dtype=float32)>
108 Use the layer to de-normalize inputs (after adapting the layer).
110 >>> adapt_data = np.array([[0., 7., 4.],
111 ... [2., 9., 6.],
112 ... [0., 7., 4.],
113 ... [2., 9., 6.]], dtype='float32')
114 >>> input_data = np.array([[1., 2., 3.]], dtype='float32')
115 >>> layer = tf.keras.layers.Normalization(axis=-1, invert=True)
116 >>> layer.adapt(adapt_data)
117 >>> layer(input_data)
118 <tf.Tensor: shape=(1, 3), dtype=float32, numpy=
119 array([2., 10., 8.], dtype=float32)>
120 """
122 def __init__(
123 self, axis=-1, mean=None, variance=None, invert=False, **kwargs
124 ):
125 super().__init__(**kwargs)
126 base_preprocessing_layer.keras_kpl_gauge.get_cell("Normalization").set(
127 True
128 )
130 # Standardize `axis` to a tuple.
131 if axis is None:
132 axis = ()
133 elif isinstance(axis, int):
134 axis = (axis,)
135 else:
136 axis = tuple(axis)
137 self.axis = axis
139 # Set `mean` and `variance` if passed.
140 if isinstance(mean, tf.Variable):
141 raise ValueError(
142 "Normalization does not support passing a Variable "
143 "for the `mean` init arg."
144 )
145 if isinstance(variance, tf.Variable):
146 raise ValueError(
147 "Normalization does not support passing a Variable "
148 "for the `variance` init arg."
149 )
150 if (mean is not None) != (variance is not None):
151 raise ValueError(
152 "When setting values directly, both `mean` and `variance` "
153 "must be set. Got mean: {} and variance: {}".format(
154 mean, variance
155 )
156 )
157 self.input_mean = mean
158 self.input_variance = variance
159 self.invert = invert
161 def build(self, input_shape):
162 super().build(input_shape)
164 if isinstance(input_shape, (list, tuple)) and all(
165 isinstance(shape, tf.TensorShape) for shape in input_shape
166 ):
167 raise ValueError(
168 "Normalization only accepts a single input. If you are "
169 "passing a python list or tuple as a single input, "
170 "please convert to a numpy array or `tf.Tensor`."
171 )
173 input_shape = tf.TensorShape(input_shape).as_list()
174 ndim = len(input_shape)
176 if any(a < -ndim or a >= ndim for a in self.axis):
177 raise ValueError(
178 "All `axis` values must be in the range [-ndim, ndim). "
179 "Found ndim: `{}`, axis: {}".format(ndim, self.axis)
180 )
182 # Axes to be kept, replacing negative values with positive equivalents.
183 # Sorted to avoid transposing axes.
184 self._keep_axis = sorted([d if d >= 0 else d + ndim for d in self.axis])
185 # All axes to be kept should have known shape.
186 for d in self._keep_axis:
187 if input_shape[d] is None:
188 raise ValueError(
189 "All `axis` values to be kept must have known shape. "
190 "Got axis: {}, "
191 "input shape: {}, with unknown axis at index: {}".format(
192 self.axis, input_shape, d
193 )
194 )
195 # Axes to be reduced.
196 self._reduce_axis = [d for d in range(ndim) if d not in self._keep_axis]
197 # 1 if an axis should be reduced, 0 otherwise.
198 self._reduce_axis_mask = [
199 0 if d in self._keep_axis else 1 for d in range(ndim)
200 ]
201 # Broadcast any reduced axes.
202 self._broadcast_shape = [
203 input_shape[d] if d in self._keep_axis else 1 for d in range(ndim)
204 ]
205 mean_and_var_shape = tuple(input_shape[d] for d in self._keep_axis)
207 if self.input_mean is None:
208 self.adapt_mean = self.add_weight(
209 name="mean",
210 shape=mean_and_var_shape,
211 dtype=self.compute_dtype,
212 initializer="zeros",
213 trainable=False,
214 )
215 self.adapt_variance = self.add_weight(
216 name="variance",
217 shape=mean_and_var_shape,
218 dtype=self.compute_dtype,
219 initializer="ones",
220 trainable=False,
221 )
222 self.count = self.add_weight(
223 name="count",
224 shape=(),
225 dtype=tf.int64,
226 initializer="zeros",
227 trainable=False,
228 )
229 self.finalize_state()
230 else:
231 # In the no adapt case, make constant tensors for mean and variance
232 # with proper broadcast shape for use during call.
233 mean = self.input_mean * np.ones(mean_and_var_shape)
234 variance = self.input_variance * np.ones(mean_and_var_shape)
235 mean = tf.reshape(mean, self._broadcast_shape)
236 variance = tf.reshape(variance, self._broadcast_shape)
237 self.mean = tf.cast(mean, self.compute_dtype)
238 self.variance = tf.cast(variance, self.compute_dtype)
240 # We override this method solely to generate a docstring.
241 def adapt(self, data, batch_size=None, steps=None):
242 """Computes the mean and variance of values in a dataset.
244 Calling `adapt()` on a `Normalization` layer is an alternative to
245 passing in `mean` and `variance` arguments during layer construction. A
246 `Normalization` layer should always either be adapted over a dataset or
247 passed `mean` and `variance`.
249 During `adapt()`, the layer will compute a `mean` and `variance`
250 separately for each position in each axis specified by the `axis`
251 argument. To calculate a single `mean` and `variance` over the input
252 data, simply pass `axis=None`.
254 In order to make `Normalization` efficient in any distribution context,
255 the computed mean and variance are kept static with respect to any
256 compiled `tf.Graph`s that call the layer. As a consequence, if the layer
257 is adapted a second time, any models using the layer should be
258 re-compiled. For more information see
259 `tf.keras.layers.experimental.preprocessing.PreprocessingLayer.adapt`.
261 `adapt()` is meant only as a single machine utility to compute layer
262 state. To analyze a dataset that cannot fit on a single machine, see
263 [Tensorflow Transform](
264 https://www.tensorflow.org/tfx/transform/get_started)
265 for a multi-machine, map-reduce solution.
267 Arguments:
268 data: The data to train on. It can be passed either as a
269 `tf.data.Dataset`, or as a numpy array.
270 batch_size: Integer or `None`.
271 Number of samples per state update.
272 If unspecified, `batch_size` will default to 32.
273 Do not specify the `batch_size` if your data is in the
274 form of datasets, generators, or `keras.utils.Sequence` instances
275 (since they generate batches).
276 steps: Integer or `None`.
277 Total number of steps (batches of samples)
278 When training with input tensors such as
279 TensorFlow data tensors, the default `None` is equal to
280 the number of samples in your dataset divided by
281 the batch size, or 1 if that cannot be determined. If x is a
282 `tf.data` dataset, and 'steps' is None, the epoch will run until
283 the input dataset is exhausted. When passing an infinitely
284 repeating dataset, you must specify the `steps` argument. This
285 argument is not supported with array inputs.
286 """
287 super().adapt(data, batch_size=batch_size, steps=steps)
289 def update_state(self, data):
290 if self.input_mean is not None:
291 raise ValueError(
292 "Cannot `adapt` a Normalization layer that is initialized with "
293 "static `mean` and `variance`, "
294 "you passed mean {} and variance {}.".format(
295 self.input_mean, self.input_variance
296 )
297 )
299 if not self.built:
300 raise RuntimeError("`build` must be called before `update_state`.")
302 data = self._standardize_inputs(data)
303 data = tf.cast(data, self.adapt_mean.dtype)
304 batch_mean, batch_variance = tf.nn.moments(data, axes=self._reduce_axis)
305 batch_shape = tf.shape(data, out_type=self.count.dtype)
306 if self._reduce_axis:
307 batch_reduce_shape = tf.gather(batch_shape, self._reduce_axis)
308 batch_count = tf.reduce_prod(batch_reduce_shape)
309 else:
310 batch_count = 1
312 total_count = batch_count + self.count
313 batch_weight = tf.cast(batch_count, dtype=self.compute_dtype) / tf.cast(
314 total_count, dtype=self.compute_dtype
315 )
316 existing_weight = 1.0 - batch_weight
318 total_mean = (
319 self.adapt_mean * existing_weight + batch_mean * batch_weight
320 )
321 # The variance is computed using the lack-of-fit sum of squares
322 # formula (see
323 # https://en.wikipedia.org/wiki/Lack-of-fit_sum_of_squares).
324 total_variance = (
325 self.adapt_variance + (self.adapt_mean - total_mean) ** 2
326 ) * existing_weight + (
327 batch_variance + (batch_mean - total_mean) ** 2
328 ) * batch_weight
329 self.adapt_mean.assign(total_mean)
330 self.adapt_variance.assign(total_variance)
331 self.count.assign(total_count)
333 def reset_state(self):
334 if self.input_mean is not None or not self.built:
335 return
337 self.adapt_mean.assign(tf.zeros_like(self.adapt_mean))
338 self.adapt_variance.assign(tf.ones_like(self.adapt_variance))
339 self.count.assign(tf.zeros_like(self.count))
341 def finalize_state(self):
342 if self.input_mean is not None or not self.built:
343 return
345 # In the adapt case, we make constant tensors for mean and variance with
346 # proper broadcast shape and dtype each time `finalize_state` is called.
347 self.mean = tf.reshape(self.adapt_mean, self._broadcast_shape)
348 self.mean = tf.cast(self.mean, self.compute_dtype)
349 self.variance = tf.reshape(self.adapt_variance, self._broadcast_shape)
350 self.variance = tf.cast(self.variance, self.compute_dtype)
352 def call(self, inputs):
353 inputs = self._standardize_inputs(inputs)
354 # The base layer automatically casts floating-point inputs, but we
355 # explicitly cast here to also allow integer inputs to be passed
356 inputs = tf.cast(inputs, self.compute_dtype)
357 if self.invert:
358 return self.mean + (
359 inputs * tf.maximum(tf.sqrt(self.variance), backend.epsilon())
360 )
361 else:
362 return (inputs - self.mean) / tf.maximum(
363 tf.sqrt(self.variance), backend.epsilon()
364 )
366 def compute_output_shape(self, input_shape):
367 return input_shape
369 def compute_output_signature(self, input_spec):
370 return input_spec
372 def get_config(self):
373 config = super().get_config()
374 config.update(
375 {
376 "axis": self.axis,
377 "invert": self.invert,
378 "mean": utils.listify_tensors(self.input_mean),
379 "variance": utils.listify_tensors(self.input_variance),
380 }
381 )
382 return config
384 def _standardize_inputs(self, inputs):
385 inputs = tf.convert_to_tensor(inputs)
386 if inputs.dtype != self.compute_dtype:
387 inputs = tf.cast(inputs, self.compute_dtype)
388 return inputs
390 def load_own_variables(self, store):
391 # Ensure that we call finalize_state after variable loading.
392 super().load_own_variables(store)
393 self.finalize_state()