Replace the apply method with the CLFProcessList which implements the LUTSequence base class and exposes a better API to apply the CLF transformation.

Author: MichaelMauderer

colour_clf_io/__init__.py

@@ -103,6 +103,13 @@
 __change_version__ = "1"
 __version__ = f"{__major_version__}.{__minor_version__}.{__change_version__}"
 
+try:
+    from colour_clf_io.processing import CLFProcessList
+
+    __all__ += ["CLFProcessList"]
+except ImportError:
+    pass
+
 
 def read_clf_from_file(path: str | Path) -> ProcessList:
     """

colour_clf_io/processing.py

@@ -2,13 +2,14 @@
 Define functionality to execute and run CLF workflows.
 """
 
-try:
-    import colour
-except ImportError:
-    raise ImportError(
+import importlib.util
+
+if importlib.util.find_spec("colour") is None:
+    err_msg = (
         "Optional dependency 'colour' not found. Please install the optional "
         "for `processing` before accessing the processing module."
     )
+    raise ImportError(err_msg)
 
 from abc import abstractmethod
 from collections.abc import Callable
@@ -32,7 +33,7 @@
 
 import colour_clf_io as clf
 
-__all__ = ["apply"]
+__all__ = ["CLFProcessList"]
 
 import numpy.typing as npt
 from colour.models.rgb.transfer_functions import (
@@ -833,48 +834,63 @@ def as_LUT_sequence_item(  # noqa: PLR0911
     raise RuntimeError(message)
 
 
-def apply(
-    process_list: clf.ProcessList,
-    value: NDArrayFloat,
-    normalised_values: bool = False,
-) -> NDArrayFloat:
+class CLFProcessList(LUTSequence):
+    """
+    Defines a *LUT* sequence created from a `colour_clf_io.ProcessList`. Creates the
+    nodes needed to execute the transformation described in the *Process List*.
     """
-    Apply the transformation described by the given :class:`colour_clf_io.ProcessList`
-    to the given value.
 
-    Parameters
-    ----------
-        process_list
-            The :class:`colour_clf_io.ProcessList` instance to apply.
-        value
-            Input value in the form of an array. Shape and data format need to be
-            compatible with the given `process_list`.
+    def __init__(
+        self,
+        process_list: clf.ProcessList,
+    ) -> None:
+        self.process_list = process_list
+        lut_sequence_items = map(as_LUT_sequence_item, process_list.process_nodes)
+        sequence = LUTSequence(*lut_sequence_items)
+        super().__init__(*sequence)
+
+    def apply(self, RGB: ArrayLike, **kwargs: Any) -> NDArrayFloat:
+        """
+        Apply the *LUT* sequence sequentially to given *RGB* colourspace
+        array.
+
+        Parameters
+        ----------
+        RGB
+            *RGB* colourspace array to apply the *LUT* sequence sequentially
+            onto.
+
+        Other Parameters
+        ----------------
         normalised_values
-            Indicates whether the input values are normalised to the range 0..1. If
-            this is the case, the range will be expanded to the input range expected
-            by the given `process_list`.
+            Argument extracted from the kwargs. Used to indicate that the
+            values passed to the apply method are already normalised.
+        kwargs
+            Keywords arguments, the keys must be the class type names for which
+            they are intended to be used with. There is no implemented way to
+            discriminate which class instance the keyword arguments should be
+            used with, thus if many class instances of the same type are
+            members of the sequence, any matching keyword arguments will be
+            used with all the class instances.
 
-    Returns
-    -------
-    :class:`NDArrayFloat`
-        Result of applying the given :class:`colour_clf_io.ProcessList`.
+        Returns
+        -------
+        :class:`numpy.ndarray`
+            Processed *RGB* colourspace array.
 
-    Raises
-    ------
-    :class:`CLFExecutionError`
-        If the given *process_list* is invalid according to the CLF specification
-        (e.g., missing or invalid parameters).
-    """
-    if not normalised_values:
-        value = value / process_list.process_nodes[0].in_bit_depth.scale_factor()
+        """
+        RGB = as_float_array(RGB)
 
-    lut_sequence_items = [
-        as_LUT_sequence_item(node) for node in process_list.process_nodes
-    ]
-    sequence = LUTSequence(*lut_sequence_items)
-    result = sequence.apply(value)
+        normalised_values = kwargs.get("normalised_values", False)
+        if not normalised_values:
+            RGB = RGB / self.process_list.process_nodes[0].in_bit_depth.scale_factor()
 
-    if not normalised_values:
-        result = result * process_list.process_nodes[-1].out_bit_depth.scale_factor()
+        result = super().apply(RGB, **kwargs)
+
+        if not normalised_values:
+            result = (
+                result
+                * self.process_list.process_nodes[-1].out_bit_depth.scale_factor()
+            )
 
-    return result
+        return result

colour_clf_io/tests/processing/test_common.py

@@ -10,7 +10,7 @@
 import numpy as np
 
 import colour_clf_io as clf
-from colour_clf_io.processing import apply
+from colour_clf_io.processing import CLFProcessList
 
 __all__ = [
     "assert_ocio_consistency",
@@ -19,6 +19,7 @@
 ]
 
 from colour.hints import NDArrayFloat
+from colour.io.luts import AbstractLUTSequenceOperator
 
 EXAMPLE_WRAPPER = """<?xml version="1.0" ?>
 <ProcessList id="Example Wrapper" compCLFversion="3.0">
@@ -66,10 +67,10 @@ def ocio_output_for_file(
     """Apply a color transform file to a flattened, one-dimensional list of
     R,G,B values.
     """
-    from PyOpenColorIO import FileTransform, GetCurrentConfig
+    import PyOpenColorIO as ocio
 
-    xform = FileTransform(src=path)
-    cpu = GetCurrentConfig().getProcessor(xform).getDefaultCPUProcessor()
+    xform = ocio.FileTransform(src=path)
+    cpu = ocio.GetCurrentConfig().getProcessor(xform).getDefaultCPUProcessor()
     result = cpu.applyRGB(rgb)
     # Note: depending on the input, `applyRGB` will either return the result data, or
     # modify the data in place. If the return value was `None` the data was modified
@@ -100,6 +101,28 @@ def result_as_array(result_text: str) -> NDArrayFloat:
     return np.array(result_values)
 
 
+class SimpleRange(AbstractLUTSequenceOperator):
+    def __init__(
+        self, in_range: tuple[float, float], out_range: tuple[float, float]
+    ) -> None:
+        self.in_range = in_range
+        self.out_range = out_range
+
+    def apply(self, x: NDArrayFloat) -> NDArrayFloat:
+        normalised = (x - self.in_range[0]) / (self.in_range[1] - self.in_range[0])
+        return normalised * (self.out_range[1] - self.out_range[0]) + self.out_range[0]
+
+
+class Normaliser(SimpleRange):
+    def __init__(self, source_range: tuple[float, float]) -> None:
+        super().__init__(source_range, (0.0, 1.0))
+
+
+class Denormaliser(SimpleRange):
+    def __init__(self, target_range: tuple[float, float]) -> None:
+        super().__init__((0.0, 1.0), target_range)
+
+
 def assert_ocio_consistency(
     value: NDArrayFloat, snippet: str, err_msg: str = "", decimals: int = 5
 ) -> None:
@@ -110,7 +133,9 @@ def assert_ocio_consistency(
     if process_list is None:
         err = "Invalid CLF snippet."
         raise AssertionError(err)
-    process_list_output = apply(process_list, value, normalised_values=True)
+    lut_sequence = CLFProcessList(process_list)
+    # apply_input_normalisation(process_list, lut_sequence)
+    process_list_output = lut_sequence.apply(value, normalised_values=True)
     value_tuple = value[0], value[1], value[2]
     ocio_output = ocio_output_for_snippet(snippet, value_tuple)
     # Note: OCIO only accepts 16-bit floats so the precision agreement is limited.
@@ -119,17 +144,19 @@ def assert_ocio_consistency(
     )
 
 
-def assert_ocio_consistency_for_file(value_rgb: NDArrayFloat, clf_path: str) -> None:
+def assert_ocio_consistency_for_file(value: NDArrayFloat, clf_path: str) -> None:
     """Assert that the colour library calculates the same output as the OCIO reference
     implementation for the given file.
     """
 
-    clf_data = clf.read_clf_from_file(clf_path)
-    if clf_data is None:
+    process_list = clf.read_clf_from_file(clf_path)
+    if process_list is None:
         err = "Invalid CLF snippet."
         raise AssertionError(err)
-    process_list_output = apply(clf_data, value_rgb, normalised_values=True)
-    ocio_output = ocio_output_for_file(clf_path, value_rgb)
+    lut_sequence = CLFProcessList(process_list)
+    # apply_input_normalisation(process_list, lut_sequence)
+    process_list_output = lut_sequence.apply(value, normalised_values=True)
+    ocio_output = ocio_output_for_file(clf_path, value)
     np.testing.assert_array_almost_equal(process_list_output, ocio_output)
 
 

colour_clf_io/tests/processing/test_helper.py

@@ -11,6 +11,7 @@
 __status__ = "Production"
 
 import numpy as np
+
 from colour_clf_io.processing import from_f16_to_uint16, from_uint16_to_f16
 
 

colour_clf_io/tests/processing/test_lut1d.py

@@ -13,8 +13,8 @@
 
 import numpy as np
 import pytest
-from colour_clf_io.processing import from_f16_to_uint16, from_uint16_to_f16
 
+from colour_clf_io.processing import from_f16_to_uint16, from_uint16_to_f16
 from colour_clf_io.tests.processing.test_common import (
     RESOURCES_ROOT,
     assert_ocio_consistency,