Add implementations for random_hue (#20620)

* Add implementations for random_hue

* Correct failed test cases

* Correct misspellings

* Update example on description

* Correct test case failed.

Author: shashaka

keras/api/_tf_keras/keras/layers/__init__.py

@@ -164,6 +164,9 @@
 from keras.src.layers.preprocessing.image_preprocessing.random_flip import (
     RandomFlip,
 )
+from keras.src.layers.preprocessing.image_preprocessing.random_hue import (
+    RandomHue,
+)
 from keras.src.layers.preprocessing.image_preprocessing.random_rotation import (
     RandomRotation,
 )

keras/api/layers/__init__.py

@@ -164,6 +164,9 @@
 from keras.src.layers.preprocessing.image_preprocessing.random_flip import (
     RandomFlip,
 )
+from keras.src.layers.preprocessing.image_preprocessing.random_hue import (
+    RandomHue,
+)
 from keras.src.layers.preprocessing.image_preprocessing.random_rotation import (
     RandomRotation,
 )

keras/src/layers/__init__.py

@@ -107,6 +107,9 @@
 from keras.src.layers.preprocessing.image_preprocessing.random_flip import (
     RandomFlip,
 )
+from keras.src.layers.preprocessing.image_preprocessing.random_hue import (
+    RandomHue,
+)
 from keras.src.layers.preprocessing.image_preprocessing.random_rotation import (
     RandomRotation,
 )

keras/src/layers/preprocessing/image_preprocessing/random_hue.py

@@ -0,0 +1,198 @@
+from keras.src.api_export import keras_export
+from keras.src.layers.preprocessing.image_preprocessing.base_image_preprocessing_layer import (  # noqa: E501
+    BaseImagePreprocessingLayer,
+)
+from keras.src.random import SeedGenerator
+
+
+def transform_value_range(images, original_range, target_range):
+    if (
+        original_range[0] == target_range[0]
+        and original_range[1] == target_range[1]
+    ):
+        return images
+
+    original_min_value, original_max_value = original_range
+    target_min_value, target_max_value = target_range
+
+    # images in the [0, 1] scale
+    images = (images - original_min_value) / (
+        original_max_value - original_min_value
+    )
+
+    scale_factor = target_max_value - target_min_value
+    return (images * scale_factor) + target_min_value
+
+
+@keras_export("keras.layers.RandomHue")
+class RandomHue(BaseImagePreprocessingLayer):
+    """Randomly adjusts the hue on given images.
+
+    This layer will randomly increase/reduce the hue for the input RGB
+    images.
+
+    The image hue is adjusted by converting the image(s) to HSV and rotating the
+    hue channel (H) by delta. The image is then converted back to RGB.
+
+    Args:
+        factor: A single float or a tuple of two floats.
+            `factor` controls the extent to which the
+            image hue is impacted. `factor=0.0` makes this layer perform a
+            no-op operation, while a value of `1.0` performs the most aggressive
+            contrast adjustment available. If a tuple is used, a `factor` is
+            sampled between the two values for every image augmented. If a
+            single float is used, a value between `0.0` and the passed float is
+            sampled. In order to ensure the value is always the same, please
+            pass a tuple with two identical floats: `(0.5, 0.5)`.
+        value_range: the range of values the incoming images will have.
+            Represented as a two-number tuple written `[low, high]`. This is
+            typically either `[0, 1]` or `[0, 255]` depending on how your
+            preprocessing pipeline is set up.
+        seed: Integer. Used to create a random seed.
+
+    ```python
+    (images, labels), _ = keras.datasets.cifar10.load_data()
+    random_hue = keras.layers.RandomHue(factor=0.5, value_range=[0, 1])
+    augmented_images = random_hue(images)
+    ```
+    """
+
+    def __init__(
+        self, factor, value_range, data_format=None, seed=None, **kwargs
+    ):
+        super().__init__(data_format=data_format, **kwargs)
+        self.factor = factor
+        self.value_range = value_range
+        self.seed = seed
+        self.generator = SeedGenerator(seed)
+
+    def parse_factor(
+        self, min_value=0.0, max_value=1.0, param_name="factor", shape=None
+    ):
+        factors = self.factor
+        if isinstance(factors, float) or isinstance(factors, int):
+            factors = (min_value, factors)
+
+        if factors[0] > factors[1]:
+            raise ValueError(
+                f"`{param_name}[0] > {param_name}[1]`, "
+                f"`{param_name}[0]` must be "
+                f"<= `{param_name}[1]`. Got `{param_name}={factors}`"
+            )
+        if (min_value is not None and factors[0] < min_value) or (
+            max_value is not None and factors[1] > max_value
+        ):
+            raise ValueError(
+                f"`{param_name}` should be inside of range "
+                f"[{min_value}, {max_value}]. Got {param_name}={factors}"
+            )
+
+        if factors[0] == factors[1]:
+            return self.backend.numpy.ones(shape=shape) * factors[0]
+
+        return self.backend.random.uniform(
+            shape,
+            seed=self.generator,
+            minval=factors[0],
+            maxval=factors[1],
+        )
+
+    def get_random_transformation(self, data, training=True, seed=None):
+        if isinstance(data, dict):
+            images = data["images"]
+        else:
+            images = data
+        images_shape = self.backend.shape(images)
+        rank = len(images_shape)
+        if rank == 3:
+            batch_size = 1
+        elif rank == 4:
+            batch_size = images_shape[0]
+        else:
+            raise ValueError(
+                "Expected the input image to be rank 3 or 4. Received "
+                f"inputs.shape={images_shape}"
+            )
+
+        if seed is None:
+            seed = self._get_seed_generator(self.backend._backend)
+        invert = self.backend.random.uniform((1,), seed=seed)
+
+        invert = self.backend.numpy.where(
+            invert > 0.5,
+            -self.backend.numpy.ones_like(invert),
+            self.backend.numpy.ones_like(invert),
+        )
+
+        factor = self.parse_factor(shape=(batch_size,))
+
+        return {"factor": invert * factor * 0.5}
+
+    def transform_images(self, images, transformation=None, training=True):
+        images = transform_value_range(images, self.value_range, (0, 1))
+        adjust_factors = transformation["factor"]
+        adjust_factors = self.backend.cast(adjust_factors, images.dtype)
+        adjust_factors = self.backend.numpy.expand_dims(adjust_factors, -1)
+        adjust_factors = self.backend.numpy.expand_dims(adjust_factors, -1)
+
+        images = self.backend.image.rgb_to_hsv(
+            images, data_format=self.data_format
+        )
+
+        if self.data_format == "channels_first":
+            h_channel = images[:, 0, :, :] + adjust_factors
+            h_channel = self.backend.numpy.where(
+                h_channel > 1.0, h_channel - 1.0, h_channel
+            )
+            h_channel = self.backend.numpy.where(
+                h_channel < 0.0, h_channel + 1.0, h_channel
+            )
+            images = self.backend.numpy.stack(
+                [h_channel, images[:, 1, :, :], images[:, 2, :, :]], axis=1
+            )
+        else:
+            h_channel = images[..., 0] + adjust_factors
+            h_channel = self.backend.numpy.where(
+                h_channel > 1.0, h_channel - 1.0, h_channel
+            )
+            h_channel = self.backend.numpy.where(
+                h_channel < 0.0, h_channel + 1.0, h_channel
+            )
+            images = self.backend.numpy.stack(
+                [h_channel, images[..., 1], images[..., 2]], axis=-1
+            )
+        images = self.backend.image.hsv_to_rgb(
+            images, data_format=self.data_format
+        )
+
+        images = self.backend.numpy.clip(images, 0, 1)
+        images = transform_value_range(images, (0, 1), self.value_range)
+
+        return images
+
+    def transform_labels(self, labels, transformation, training=True):
+        return labels
+
+    def transform_segmentation_masks(
+        self, segmentation_masks, transformation, training=True
+    ):
+        return segmentation_masks
+
+    def transform_bounding_boxes(
+        self, bounding_boxes, transformation, training=True
+    ):
+        return bounding_boxes
+
+    def get_config(self):
+        config = super().get_config()
+        config.update(
+            {
+                "factor": self.factor,
+                "value_range": self.value_range,
+                "seed": self.seed,
+            }
+        )
+        return config
+
+    def compute_output_shape(self, input_shape):
+        return input_shape

keras/src/layers/preprocessing/image_preprocessing/random_hue_test.py

@@ -0,0 +1,66 @@
+import numpy as np
+import pytest
+from tensorflow import data as tf_data
+
+import keras
+from keras.src import backend
+from keras.src import layers
+from keras.src import testing
+
+
+class RandomHueTest(testing.TestCase):
+    @pytest.mark.requires_trainable_backend
+    def test_layer(self):
+        self.run_layer_test(
+            layers.RandomHue,
+            init_kwargs={
+                "factor": 0.75,
+                "value_range": (20, 200),
+                "seed": 1,
+            },
+            input_shape=(8, 3, 4, 3),
+            supports_masking=False,
+            expected_output_shape=(8, 3, 4, 3),
+        )
+
+    def test_random_hue_value_range(self):
+        image = keras.random.uniform(shape=(3, 3, 3), minval=0, maxval=1)
+
+        layer = layers.RandomHue(0.2, (0, 255))
+        adjusted_image = layer(image)
+
+        self.assertTrue(keras.ops.numpy.all(adjusted_image >= 0))
+        self.assertTrue(keras.ops.numpy.all(adjusted_image <= 1))
+
+    def test_random_hue_no_change_with_zero_factor(self):
+        data_format = backend.config.image_data_format()
+        if data_format == "channels_last":
+            inputs = keras.random.randint((224, 224, 3), 0, 255)
+        else:
+            inputs = keras.random.randint((3, 224, 224), 0, 255)
+
+        layer = layers.RandomHue(0, (0, 255), data_format=data_format)
+        output = layer(inputs, training=False)
+        self.assertAllClose(inputs, output, atol=1e-3, rtol=1e-5)
+
+    def test_random_hue_randomness(self):
+        image = keras.random.uniform(shape=(3, 3, 3), minval=0, maxval=1)[:5]
+
+        layer = layers.RandomHue(0.2, (0, 255))
+        adjusted_images = layer(image)
+
+        self.assertNotAllClose(adjusted_images, image)
+
+    def test_tf_data_compatibility(self):
+        data_format = backend.config.image_data_format()
+        if data_format == "channels_last":
+            input_data = np.random.random((2, 8, 8, 3))
+        else:
+            input_data = np.random.random((2, 3, 8, 8))
+        layer = layers.RandomHue(
+            factor=0.5, value_range=[0, 1], data_format=data_format, seed=1337
+        )
+
+        ds = tf_data.Dataset.from_tensor_slices(input_data).batch(2).map(layer)
+        for output in ds.take(1):
+            output.numpy()