Source code for colour.continuous.signal

"""
Signal
======

Defines the class implementing support for continuous signal:

-   :class:`colour.continuous.Signal`
"""

from __future__ import annotations

from collections.abc import Iterator, Mapping, Sequence, ValuesView
from operator import (
    add,
    iadd,
    imul,
    ipow,
    isub,
    itruediv,
    mul,
    pow,
    sub,
    truediv,
)

import numpy as np

from colour.algebra import Extrapolator, KernelInterpolator
from colour.constants import DTYPE_FLOAT_DEFAULT
from colour.continuous import AbstractContinuousFunction
from colour.hints import (
    TYPE_CHECKING,
    Any,
    ArrayLike,
    Callable,
    DTypeFloat,
    Literal,
    NDArrayFloat,
    Optional,
    ProtocolExtrapolator,
    ProtocolInterpolator,
    Real,
    Self,
    Type,
    Union,
    cast,
)
from colour.utilities import (
    as_float_array,
    attest,
    fill_nan,
    full,
    is_pandas_installed,
    multiline_repr,
    ndarray_copy,
    ndarray_copy_enable,
    optional,
    required,
    runtime_warning,
    tsplit,
    tstack,
    validate_method,
)
from colour.utilities.common import int_digest
from colour.utilities.documentation import is_documentation_building

if TYPE_CHECKING or is_pandas_installed():
    from pandas import Series  # pragma: no cover
else:  # pragma: no cover
    from unittest import mock

    Series = mock.MagicMock()

__author__ = "Colour Developers"
__copyright__ = "Copyright 2013 Colour Developers"
__license__ = "BSD-3-Clause - https://opensource.org/licenses/BSD-3-Clause"
__maintainer__ = "Colour Developers"
__email__ = "colour-developers@colour-science.org"
__status__ = "Production"

__all__ = [
    "Signal",
]


[docs] class Signal(AbstractContinuousFunction): """ Define the base class for continuous signal. The class implements the :meth:`Signal.function` method so that evaluating the function for any independent domain variable :math:`x \\in\\mathbb{R}` returns a corresponding range variable :math:`y \\in\\mathbb{R}`. It adopts an interpolating function encapsulated inside an extrapolating function. The resulting function independent domain, stored as discrete values in the :attr:`colour.continuous.Signal.domain` property corresponds with the function dependent and already known range stored in the :attr:`colour.continuous.Signal.range` property. .. important:: Specific documentation about getting, setting, indexing and slicing the continuous signal values is available in the :ref:`spectral-representation-and-continuous-signal` section. Parameters ---------- data Data to be stored in the continuous signal. domain Values to initialise the :attr:`colour.continuous.Signal.domain` attribute with. If both ``data`` and ``domain`` arguments are defined, the latter with be used to initialise the :attr:`colour.continuous.Signal.domain` property. Other Parameters ---------------- dtype float point data type. extrapolator Extrapolator class type to use as extrapolating function. extrapolator_kwargs Arguments to use when instantiating the extrapolating function. interpolator Interpolator class type to use as interpolating function. interpolator_kwargs Arguments to use when instantiating the interpolating function. name Continuous signal name. Attributes ---------- - :attr:`~colour.continuous.Signal.dtype` - :attr:`~colour.continuous.Signal.domain` - :attr:`~colour.continuous.Signal.range` - :attr:`~colour.continuous.Signal.interpolator` - :attr:`~colour.continuous.Signal.interpolator_kwargs` - :attr:`~colour.continuous.Signal.extrapolator` - :attr:`~colour.continuous.Signal.extrapolator_kwargs` - :attr:`~colour.continuous.Signal.function` Methods ------- - :meth:`~colour.continuous.Signal.__init__` - :meth:`~colour.continuous.Signal.__str__` - :meth:`~colour.continuous.Signal.__repr__` - :meth:`~colour.continuous.Signal.__hash__` - :meth:`~colour.continuous.Signal.__getitem__` - :meth:`~colour.continuous.Signal.__setitem__` - :meth:`~colour.continuous.Signal.__contains__` - :meth:`~colour.continuous.Signal.__eq__` - :meth:`~colour.continuous.Signal.__ne__` - :meth:`~colour.continuous.Signal.arithmetical_operation` - :meth:`~colour.continuous.Signal.signal_unpack_data` - :meth:`~colour.continuous.Signal.fill_nan` - :meth:`~colour.continuous.Signal.to_series` Examples -------- Instantiation with implicit *domain*: >>> range_ = np.linspace(10, 100, 10) >>> print(Signal(range_)) [[ 0. 10.] [ 1. 20.] [ 2. 30.] [ 3. 40.] [ 4. 50.] [ 5. 60.] [ 6. 70.] [ 7. 80.] [ 8. 90.] [ 9. 100.]] Instantiation with explicit *domain*: >>> domain = np.arange(100, 1100, 100) >>> print(Signal(range_, domain)) [[ 100. 10.] [ 200. 20.] [ 300. 30.] [ 400. 40.] [ 500. 50.] [ 600. 60.] [ 700. 70.] [ 800. 80.] [ 900. 90.] [ 1000. 100.]] Instantiation with a *dict*: >>> print(Signal(dict(zip(domain, range_)))) [[ 100. 10.] [ 200. 20.] [ 300. 30.] [ 400. 40.] [ 500. 50.] [ 600. 60.] [ 700. 70.] [ 800. 80.] [ 900. 90.] [ 1000. 100.]] Instantiation with a *Pandas* :class:`pandas.Series`: >>> if is_pandas_installed(): ... from pandas import Series ... ... print(Signal(Series(dict(zip(domain, range_))))) # doctest: +SKIP [[ 100. 10.] [ 200. 20.] [ 300. 30.] [ 400. 40.] [ 500. 50.] [ 600. 60.] [ 700. 70.] [ 800. 80.] [ 900. 90.] [ 1000. 100.]] Retrieving domain *y* variable for arbitrary range *x* variable: >>> x = 150 >>> range_ = np.sin(np.linspace(0, 1, 10)) >>> Signal(range_, domain)[x] # doctest: +ELLIPSIS 0.0359701... >>> x = np.linspace(100, 1000, 3) >>> Signal(range_, domain)[x] # doctest: +ELLIPSIS array([ ..., 4.7669395...e-01, 8.4147098...e-01]) Using an alternative interpolating function: >>> x = 150 >>> from colour.algebra import CubicSplineInterpolator >>> Signal(range_, domain, interpolator=CubicSplineInterpolator)[ ... x ... ] # doctest: +ELLIPSIS 0.0555274... >>> x = np.linspace(100, 1000, 3) >>> Signal(range_, domain, interpolator=CubicSplineInterpolator)[ ... x ... ] # doctest: +ELLIPSIS array([ 0. , 0.4794253..., 0.8414709...]) """
[docs] def __init__( self, data: ArrayLike | dict | Self | Series | None = None, domain: ArrayLike | None = None, **kwargs: Any, ) -> None: super().__init__(kwargs.get("name")) self._dtype: Type[DTypeFloat] = DTYPE_FLOAT_DEFAULT self._domain: NDArrayFloat = np.array([]) self._range: NDArrayFloat = np.array([]) self._interpolator: Type[ProtocolInterpolator] = KernelInterpolator self._interpolator_kwargs: dict = {} self._extrapolator: Type[ProtocolExtrapolator] = Extrapolator self._extrapolator_kwargs: dict = { "method": "Constant", "left": np.nan, "right": np.nan, } self.range, self.domain = self.signal_unpack_data(data, domain)[::-1] self.dtype = kwargs.get("dtype", self._dtype) self.interpolator = kwargs.get("interpolator", self._interpolator) self.interpolator_kwargs = kwargs.get( "interpolator_kwargs", self._interpolator_kwargs ) self.extrapolator = kwargs.get("extrapolator", self._extrapolator) self.extrapolator_kwargs = kwargs.get( "extrapolator_kwargs", self._extrapolator_kwargs ) self._function: Callable | None = None
@property def dtype(self) -> Type[DTypeFloat]: """ Getter and setter property for the continuous signal dtype. Parameters ---------- value Value to set the continuous signal dtype with. Returns ------- DTypeFloat Continuous signal dtype. """ return self._dtype @dtype.setter def dtype(self, value: Type[DTypeFloat]): """Setter for the **self.dtype** property.""" attest( value in DTypeFloat.__args__, # pyright: ignore f'"dtype" must be one of the following types: {DTypeFloat.__args__}', # pyright: ignore ) self._dtype = value # The following self-assignments are written as intended and # triggers the rebuild of the underlying function. if self.domain.dtype != value or self.range.dtype != value: self.domain = self.domain self.range = self.range @property def domain(self) -> NDArrayFloat: """ Getter and setter property for the continuous signal independent domain variable :math:`x`. Parameters ---------- value Value to set the continuous signal independent domain variable :math:`x` with. Returns ------- :class:`numpy.ndarray` Continuous signal independent domain variable :math:`x`. """ return ndarray_copy(self._domain) @domain.setter def domain(self, value: ArrayLike): """Setter for the **self.domain** property.""" value = as_float_array(value, self.dtype) if not np.all(np.isfinite(value)): runtime_warning( f'"{self.name}" new "domain" variable is not finite: {value}, ' f"unpredictable results may occur!" ) else: attest( np.all(value[:-1] <= value[1:]), "The new domain value is not monotonic! ", ) if value.size != self._range.size: self._range = np.resize(self._range, value.shape) self._domain = value self._function = None # Invalidate the underlying continuous function. @property def range(self) -> NDArrayFloat: """ Getter and setter property for the continuous signal corresponding range variable :math:`y`. Parameters ---------- value Value to set the continuous signal corresponding range :math:`y` variable with. Returns ------- :class:`numpy.ndarray` Continuous signal corresponding range variable :math:`y`. """ return ndarray_copy(self._range) @range.setter def range(self, value: ArrayLike): """Setter for the **self.range** property.""" value = as_float_array(value, self.dtype) if not np.all(np.isfinite(value)): runtime_warning( f'"{self.name}" new "range" variable is not finite: {value}, ' f"unpredictable results may occur!" ) # Empty domain occurs during __init__ because range is set before domain attest( self._domain.size in (0, self._domain.size), '"domain" and "range" variables must have same size!', ) self._range = value self._function = None # Invalidate the underlying continuous function. @property def interpolator(self) -> Type[ProtocolInterpolator]: """ Getter and setter property for the continuous signal interpolator type. Parameters ---------- value Value to set the continuous signal interpolator type with. Returns ------- Type[ProtocolInterpolator] Continuous signal interpolator type. """ return self._interpolator @interpolator.setter def interpolator(self, value: Type[ProtocolInterpolator]): """Setter for the **self.interpolator** property.""" # TODO: Check for interpolator compatibility. self._interpolator = value self._function = None # Invalidate the underlying continuous function. @property def interpolator_kwargs(self) -> dict: """ Getter and setter property for the continuous signal interpolator instantiation time arguments. Parameters ---------- value Value to set the continuous signal interpolator instantiation time arguments to. Returns ------- :class:`dict` Continuous signal interpolator instantiation time arguments. """ return self._interpolator_kwargs @interpolator_kwargs.setter def interpolator_kwargs(self, value: dict): """Setter for the **self.interpolator_kwargs** property.""" attest( isinstance(value, dict), f'"interpolator_kwargs" property: "{value}" type is not "dict"!', ) self._interpolator_kwargs = value self._function = None # Invalidate the underlying continuous function. @property def extrapolator(self) -> Type[ProtocolExtrapolator]: """ Getter and setter property for the continuous signal extrapolator type. Parameters ---------- value Value to set the continuous signal extrapolator type with. Returns ------- Type[ProtocolExtrapolator] Continuous signal extrapolator type. """ return self._extrapolator @extrapolator.setter def extrapolator(self, value: Type[ProtocolExtrapolator]): """Setter for the **self.extrapolator** property.""" # TODO: Check for extrapolator compatibility. self._extrapolator = value self._function = None # Invalidate the underlying continuous function. @property def extrapolator_kwargs(self) -> dict: """ Getter and setter property for the continuous signal extrapolator instantiation time arguments. Parameters ---------- value Value to set the continuous signal extrapolator instantiation time arguments to. Returns ------- :class:`dict` Continuous signal extrapolator instantiation time arguments. """ return self._extrapolator_kwargs @extrapolator_kwargs.setter def extrapolator_kwargs(self, value: dict): """Setter for the **self.extrapolator_kwargs** property.""" attest( isinstance(value, dict), f'"extrapolator_kwargs" property: "{value}" type is not "dict"!', ) self._extrapolator_kwargs = value self._function = None # Invalidate the underlying continuous function. @property @ndarray_copy_enable(False) def function(self) -> Callable: """ Getter property for the continuous signal callable. Returns ------- Callable Continuous signal callable. """ if self._function is None: # Create the underlying continuous function. if self._domain.size != 0 and self._range.size != 0: self._function = self._extrapolator( self._interpolator( self._domain, self._range, **self._interpolator_kwargs ), **self._extrapolator_kwargs, ) else: def _undefined_function( *args: Any, # noqa: ARG001 **kwargs: Any, # noqa: ARG001 ): """ Raise a :class:`ValueError` exception. Other Parameters ---------------- args Arguments. kwargs Keywords arguments. Raises ------ ValueError """ raise ValueError( "Underlying signal interpolator function does not " 'exists, please ensure that both "domain" and "range" ' "variables are defined!" ) self._function = cast(Callable, _undefined_function) return cast(Callable, self._function)
[docs] @ndarray_copy_enable(False) def __str__(self) -> str: """ Return a formatted string representation of the continuous signal. Returns ------- :class:`str` Formatted string representation. Examples -------- >>> range_ = np.linspace(10, 100, 10) >>> print(Signal(range_)) [[ 0. 10.] [ 1. 20.] [ 2. 30.] [ 3. 40.] [ 4. 50.] [ 5. 60.] [ 6. 70.] [ 7. 80.] [ 8. 90.] [ 9. 100.]] """ return str(tstack([self._domain, self._range]))
[docs] @ndarray_copy_enable(False) def __repr__(self) -> str: """ Return an evaluable string representation of the continuous signal. Returns ------- :class:`str` Evaluable string representation. Examples -------- >>> range_ = np.linspace(10, 100, 10) >>> Signal(range_) Signal([[ 0., 10.], [ 1., 20.], [ 2., 30.], [ 3., 40.], [ 4., 50.], [ 5., 60.], [ 6., 70.], [ 7., 80.], [ 8., 90.], [ 9., 100.]], KernelInterpolator, {}, Extrapolator, {'method': 'Constant', 'left': nan, 'right': nan}) """ if is_documentation_building(): # pragma: no cover return f"{self.__class__.__name__}(name='{self.name}', ...)" return multiline_repr( self, [ { "formatter": lambda x: repr( # noqa: ARG005 tstack([self._domain, self._range]) ), }, { "name": "interpolator", "formatter": lambda x: self._interpolator.__name__, # noqa: ARG005 }, {"name": "interpolator_kwargs"}, { "name": "extrapolator", "formatter": lambda x: self._extrapolator.__name__, # noqa: ARG005 }, {"name": "extrapolator_kwargs"}, ], )
[docs] @ndarray_copy_enable(False) def __hash__(self) -> int: """ Return the abstract continuous function hash. Returns ------- :class:`int` Object hash. """ return hash( ( int_digest(self._domain.tobytes()), int_digest(self._range.tobytes()), self.interpolator.__name__, repr(self.interpolator_kwargs), self.extrapolator.__name__, repr(self.extrapolator_kwargs), ) )
[docs] def __getitem__(self, x: ArrayLike | slice) -> NDArrayFloat: """ Return the corresponding range variable :math:`y` for independent domain variable :math:`x`. Parameters ---------- x Independent domain variable :math:`x`. Returns ------- :class:`numpy.ndarray` Variable :math:`y` range value. Examples -------- >>> range_ = np.linspace(10, 100, 10) >>> signal = Signal(range_) >>> print(signal) [[ 0. 10.] [ 1. 20.] [ 2. 30.] [ 3. 40.] [ 4. 50.] [ 5. 60.] [ 6. 70.] [ 7. 80.] [ 8. 90.] [ 9. 100.]] >>> signal[0] 10.0 >>> signal[np.array([0, 1, 2])] array([ 10., 20., 30.]) >>> signal[0:3] array([ 10., 20., 30.]) >>> signal[np.linspace(0, 5, 5)] # doctest: +ELLIPSIS array([ 10. , 22.8348902..., 34.8004492..., \ 47.5535392..., 60. ]) """ if isinstance(x, slice): return self._range[x] else: return self.function(x)
[docs] def __setitem__(self, x: ArrayLike | slice, y: ArrayLike): """ Set the corresponding range variable :math:`y` for independent domain variable :math:`x`. Parameters ---------- x Independent domain variable :math:`x`. y Corresponding range variable :math:`y`. Examples -------- >>> range_ = np.linspace(10, 100, 10) >>> signal = Signal(range_) >>> print(signal) [[ 0. 10.] [ 1. 20.] [ 2. 30.] [ 3. 40.] [ 4. 50.] [ 5. 60.] [ 6. 70.] [ 7. 80.] [ 8. 90.] [ 9. 100.]] >>> signal[0] = 20 >>> signal[0] 20.0 >>> signal[np.array([0, 1, 2])] = 30 >>> signal[np.array([0, 1, 2])] array([ 30., 30., 30.]) >>> signal[0:3] = 40 >>> signal[0:3] array([ 40., 40., 40.]) >>> signal[np.linspace(0, 5, 5)] = 50 >>> print(signal) [[ 0. 50. ] [ 1. 40. ] [ 1.25 50. ] [ 2. 40. ] [ 2.5 50. ] [ 3. 40. ] [ 3.75 50. ] [ 4. 50. ] [ 5. 50. ] [ 6. 70. ] [ 7. 80. ] [ 8. 90. ] [ 9. 100. ]] >>> signal[np.array([0, 1, 2])] = np.array([10, 20, 30]) >>> print(signal) [[ 0. 10. ] [ 1. 20. ] [ 1.25 50. ] [ 2. 30. ] [ 2.5 50. ] [ 3. 40. ] [ 3.75 50. ] [ 4. 50. ] [ 5. 50. ] [ 6. 70. ] [ 7. 80. ] [ 8. 90. ] [ 9. 100. ]] """ if isinstance(x, slice): self._range[x] = y else: x = np.atleast_1d(x).astype(self.dtype) y = np.resize(y, x.shape) # Matching domain, updating existing `self._range` values. mask = np.in1d(x, self._domain) x_m = x[mask] indexes = np.searchsorted(self._domain, x_m) self._range[indexes] = y[mask] # Non matching domain, inserting into existing `self.domain` # and `self.range`. x_nm = x[~mask] indexes = np.searchsorted(self._domain, x_nm) if indexes.size != 0: self._domain = np.insert(self._domain, indexes, x_nm) self._range = np.insert(self._range, indexes, y[~mask]) self._function = None # Invalidate the underlying continuous function.
[docs] def __contains__(self, x: ArrayLike | slice) -> bool: """ Return whether the continuous signal contains given independent domain variable :math:`x`. Parameters ---------- x Independent domain variable :math:`x`. Returns ------- :class:`bool` Whether :math:`x` domain value is contained. Examples -------- >>> range_ = np.linspace(10, 100, 10) >>> signal = Signal(range_) >>> 0 in signal True >>> 0.5 in signal True >>> 1000 in signal False """ return bool( np.all( np.where( np.logical_and( x >= np.min(self._domain), # pyright: ignore x <= np.max(self._domain), # pyright: ignore ), True, False, ) ) )
[docs] @ndarray_copy_enable(False) def __eq__(self, other: Any) -> bool: """ Return whether the continuous signal is equal to given other object. Parameters ---------- other Object to test whether it is equal to the continuous signal. Returns ------- :class:`bool` Whether given object is equal to the continuous signal. Examples -------- >>> range_ = np.linspace(10, 100, 10) >>> signal_1 = Signal(range_) >>> signal_2 = Signal(range_) >>> signal_1 == signal_2 True >>> signal_2[0] = 20 >>> signal_1 == signal_2 False >>> signal_2[0] = 10 >>> signal_1 == signal_2 True >>> from colour.algebra import CubicSplineInterpolator >>> signal_2.interpolator = CubicSplineInterpolator >>> signal_1 == signal_2 False """ # NOTE: Comparing "interpolator_kwargs" and "extrapolator_kwargs" using # their string representation because of presence of NaNs. if isinstance(other, Signal): return all( [ np.array_equal(self._domain, other.domain), np.array_equal(self._range, other.range), self._interpolator is other.interpolator, repr(self._interpolator_kwargs) == repr(other.interpolator_kwargs), self._extrapolator is other.extrapolator, repr(self._extrapolator_kwargs) == repr(other.extrapolator_kwargs), ] ) else: return False
[docs] def __ne__(self, other: Any) -> bool: """ Return whether the continuous signal is not equal to given other object. Parameters ---------- other Object to test whether it is not equal to the continuous signal. Returns ------- :class:`bool` Whether given object is not equal to the continuous signal. Examples -------- >>> range_ = np.linspace(10, 100, 10) >>> signal_1 = Signal(range_) >>> signal_2 = Signal(range_) >>> signal_1 != signal_2 False >>> signal_2[0] = 20 >>> signal_1 != signal_2 True >>> signal_2[0] = 10 >>> signal_1 != signal_2 False >>> from colour.algebra import CubicSplineInterpolator >>> signal_2.interpolator = CubicSplineInterpolator >>> signal_1 != signal_2 True """ return not (self == other)
@ndarray_copy_enable(False) def _fill_domain_nan( self, method: Literal["Constant", "Interpolation"] | str = "Interpolation", default: Real = 0, ): """ Fill NaNs in independent domain variable :math:`x` using given method. Parameters ---------- method *Interpolation* method linearly interpolates through the NaNs, *Constant* method replaces NaNs with ``default``. default Value to use with the *Constant* method. Returns ------- :class:`colour.continuous.Signal` NaNs filled continuous signal independent domain :math:`x` variable. """ self.domain = fill_nan(self._domain, method, default) @ndarray_copy_enable(False) def _fill_range_nan( self, method: Literal["Constant", "Interpolation"] | str = "Interpolation", default: Real = 0, ): """ Fill NaNs in corresponding range variable :math:`y` using given method. Parameters ---------- method *Interpolation* method linearly interpolates through the NaNs, *Constant* method replaces NaNs with ``default``. default Value to use with the *Constant* method. Returns ------- :class:`colour.continuous.Signal` NaNs filled continuous signal i corresponding range :math:`y` variable. """ self.range = fill_nan(self._range, method, default)
[docs] @ndarray_copy_enable(False) def arithmetical_operation( self, a: ArrayLike | AbstractContinuousFunction, operation: Literal["+", "-", "*", "/", "**"], in_place: bool = False, ) -> AbstractContinuousFunction: """ Perform given arithmetical operation with operand :math:`a`, the operation can be either performed on a copy or in-place. Parameters ---------- a Operand :math:`a`. operation Operation to perform. in_place Operation happens in place. Returns ------- :class:`colour.continuous.Signal` Continuous signal. Examples -------- Adding a single *numeric* variable: >>> range_ = np.linspace(10, 100, 10) >>> signal_1 = Signal(range_) >>> print(signal_1) [[ 0. 10.] [ 1. 20.] [ 2. 30.] [ 3. 40.] [ 4. 50.] [ 5. 60.] [ 6. 70.] [ 7. 80.] [ 8. 90.] [ 9. 100.]] >>> print(signal_1.arithmetical_operation(10, "+", True)) [[ 0. 20.] [ 1. 30.] [ 2. 40.] [ 3. 50.] [ 4. 60.] [ 5. 70.] [ 6. 80.] [ 7. 90.] [ 8. 100.] [ 9. 110.]] Adding an `ArrayLike` variable: >>> a = np.linspace(10, 100, 10) >>> print(signal_1.arithmetical_operation(a, "+", True)) [[ 0. 30.] [ 1. 50.] [ 2. 70.] [ 3. 90.] [ 4. 110.] [ 5. 130.] [ 6. 150.] [ 7. 170.] [ 8. 190.] [ 9. 210.]] Adding a :class:`colour.continuous.Signal` class: >>> signal_2 = Signal(range_) >>> print(signal_1.arithmetical_operation(signal_2, "+", True)) [[ 0. 40.] [ 1. 70.] [ 2. 100.] [ 3. 130.] [ 4. 160.] [ 5. 190.] [ 6. 220.] [ 7. 250.] [ 8. 280.] [ 9. 310.]] """ operator, ioperator = { "+": (add, iadd), "-": (sub, isub), "*": (mul, imul), "/": (truediv, itruediv), "**": (pow, ipow), }[operation] if in_place: if isinstance(a, Signal): self[self._domain] = operator(self._range, a[self._domain]) exclusive_or = np.setxor1d(self._domain, a.domain) self[exclusive_or] = full(exclusive_or.shape, np.nan) else: self.range = ioperator(self._range, a) return self else: copy = ioperator(self.copy(), a) return copy
[docs] @staticmethod @ndarray_copy_enable(True) def signal_unpack_data( data=Optional[Union[ArrayLike, dict, Series, "Signal"]], domain: ArrayLike | None = None, dtype: Type[DTypeFloat] | None = None, ) -> tuple: """ Unpack given data for continuous signal instantiation. Parameters ---------- data Data to unpack for continuous signal instantiation. domain Values to initialise the :attr:`colour.continuous.Signal.domain` attribute with. If both ``data`` and ``domain`` arguments are defined, the latter will be used to initialise the :attr:`colour.continuous.Signal.domain` property. dtype float point data type. Returns ------- :class:`tuple` Independent domain variable :math:`x` and corresponding range variable :math:`y` unpacked for continuous signal instantiation. Examples -------- Unpacking using implicit *domain*: >>> range_ = np.linspace(10, 100, 10) >>> domain, range_ = Signal.signal_unpack_data(range_) >>> print(domain) [ 0. 1. 2. 3. 4. 5. 6. 7. 8. 9.] >>> print(range_) [ 10. 20. 30. 40. 50. 60. 70. 80. 90. 100.] Unpacking using explicit *domain*: >>> domain = np.arange(100, 1100, 100) >>> domain, range = Signal.signal_unpack_data(range_, domain) >>> print(domain) [ 100. 200. 300. 400. 500. 600. 700. 800. 900. 1000.] >>> print(range_) [ 10. 20. 30. 40. 50. 60. 70. 80. 90. 100.] Unpacking using a *dict*: >>> domain, range_ = Signal.signal_unpack_data(dict(zip(domain, range_))) >>> print(domain) [ 100. 200. 300. 400. 500. 600. 700. 800. 900. 1000.] >>> print(range_) [ 10. 20. 30. 40. 50. 60. 70. 80. 90. 100.] Unpacking using a *Pandas* :class:`pandas.Series`: >>> if is_pandas_installed(): ... from pandas import Series ... ... domain, range = Signal.signal_unpack_data( ... Series(dict(zip(domain, range_))) ... ) ... # doctest: +ELLIPSIS >>> print(domain) # doctest: +SKIP [ 100. 200. 300. 400. 500. 600. 700. 800. 900. 1000.] >>> print(range_) # doctest: +SKIP [ 10. 20. 30. 40. 50. 60. 70. 80. 90. 100.] Unpacking using a :class:`colour.continuous.Signal` class: >>> domain, range_ = Signal.signal_unpack_data(Signal(range_, domain)) >>> print(domain) [ 100. 200. 300. 400. 500. 600. 700. 800. 900. 1000.] >>> print(range_) [ 10. 20. 30. 40. 50. 60. 70. 80. 90. 100.] """ dtype = optional(dtype, DTYPE_FLOAT_DEFAULT) domain_unpacked: NDArrayFloat = np.array([]) range_unpacked: NDArrayFloat = np.array([]) if isinstance(data, Signal): domain_unpacked = data.domain range_unpacked = data.range elif issubclass(type(data), Sequence) or isinstance( data, (tuple, list, np.ndarray, Iterator, ValuesView) ): data_array = ( tsplit(list(cast(Sequence, data))) if not isinstance(data, np.ndarray) else data ) attest(data_array.ndim == 1, 'User "data" must be 1-dimensional!') domain_unpacked, range_unpacked = ( np.arange(0, data_array.size, dtype=dtype), data_array, ) elif issubclass(type(data), Mapping) or isinstance(data, dict): domain_unpacked, range_unpacked = tsplit( sorted(cast(Mapping, data).items()) ) elif is_pandas_installed() and isinstance(data, Series): domain_unpacked = data.index.values range_unpacked = data.values if domain is not None: domain_array = as_float_array(domain, dtype) attest( len(domain_array) == len(range_unpacked), 'User "domain" length is not compatible with unpacked "range"!', ) domain_unpacked = domain_array if range_unpacked is not None: range_unpacked = as_float_array(range_unpacked, dtype) return ndarray_copy(domain_unpacked), ndarray_copy(range_unpacked)
[docs] def fill_nan( self, method: Literal["Constant", "Interpolation"] | str = "Interpolation", default: Real = 0, ) -> Signal: """ Fill NaNs in independent domain variable :math:`x` and corresponding range variable :math:`y` using given method. Parameters ---------- method *Interpolation* method linearly interpolates through the NaNs, *Constant* method replaces NaNs with ``default``. default Value to use with the *Constant* method. Returns ------- :class:`colour.continuous.Signal` NaNs filled continuous signal. Examples -------- >>> range_ = np.linspace(10, 100, 10) >>> signal = Signal(range_) >>> signal[3:7] = np.nan >>> print(signal) [[ 0. 10.] [ 1. 20.] [ 2. 30.] [ 3. nan] [ 4. nan] [ 5. nan] [ 6. nan] [ 7. 80.] [ 8. 90.] [ 9. 100.]] >>> print(signal.fill_nan()) [[ 0. 10.] [ 1. 20.] [ 2. 30.] [ 3. 40.] [ 4. 50.] [ 5. 60.] [ 6. 70.] [ 7. 80.] [ 8. 90.] [ 9. 100.]] >>> signal[3:7] = np.nan >>> print(signal.fill_nan(method="Constant")) [[ 0. 10.] [ 1. 20.] [ 2. 30.] [ 3. 0.] [ 4. 0.] [ 5. 0.] [ 6. 0.] [ 7. 80.] [ 8. 90.] [ 9. 100.]] """ method = validate_method(method, ("Interpolation", "Constant")) self._fill_domain_nan(method, default) self._fill_range_nan(method, default) return self
[docs] @required("Pandas") def to_series(self) -> Series: """ Convert the continuous signal to a *Pandas* :class:`pandas.Series` class instance. Returns ------- :class:`pandas.Series` Continuous signal as a *Pandas*:class:`pandas.Series` class instance. Examples -------- >>> if is_pandas_installed(): ... range_ = np.linspace(10, 100, 10) ... signal = Signal(range_) ... print(signal.to_series()) # doctest: +SKIP 0.0 10.0 1.0 20.0 2.0 30.0 3.0 40.0 4.0 50.0 5.0 60.0 6.0 70.0 7.0 80.0 8.0 90.0 9.0 100.0 Name: Signal (...), dtype: float64 """ return Series(data=self._range, index=self._domain, name=self.name)