API reference

earthcarekit.utils.numpy

Utilities based on numpy.

Notes

This module does not depend on other internal modules.

---

all_in

all_in(subset: ArrayLike, set: ArrayLike) -> bool

Check if all elements in subset are present in set.

Parameters:

Name	Type	Description	Default
subset	ArrayLike	The list to check.	required
set	ArrayLike	The list to check against.	required

Returns:

Type	Description
bool	True if all elements of subset are in set, False otherwise.

Source code in earthcarekit/utils/numpy/_check.py

def all_in(subset: ArrayLike, set: ArrayLike) -> bool:
    """Check if all elements in `subset` are present in `set`.

    Args:
        subset: The list to check.
        set: The list to check against.

    Returns:
        True if all elements of `subset` are in `set`, False otherwise.
    """
    return all(item in np.asarray(set) for item in np.asarray(subset))

all_same

all_same(a: ArrayLike) -> bool

Check if all elements in the input array are the same.

Parameters:

Name	Type	Description	Default
a	ArrayLike	Input array or array-like object to check.	required

Returns:

Type	Description
bool	True if all elements in the array are the same, False otherwise.

Source code in earthcarekit/utils/numpy/_check.py

def all_same(a: ArrayLike) -> bool:
    """Check if all elements in the input array are the same.

    Args:
        a: Input array or array-like object to check.

    Returns:
        True if all elements in the array are the same, False otherwise.
    """
    a = np.asarray(a)
    return bool(np.all(a == a[0]))

asarray_or_none

asarray_or_none(a: None, **kwargs) -> None

asarray_or_none(a: ArrayLike, **kwargs) -> NDArray

asarray_or_none(a, **kwargs)

Convert input to numpy.ndarray unless it is None.

Source code in earthcarekit/utils/numpy/_asarray_or_none.py

def asarray_or_none(a, **kwargs):
    """Convert input to `numpy.ndarray` unless it is None."""
    if a is None:
        return None
    return np.asarray(a, **kwargs)

bins_to_centers

bins_to_centers(bins: ArrayLike) -> NDArray

Converts bin edges to bin centers.

Parameters:

Name	Type	Description	Default
bins	ArrayLike	Array of N+1 bin edges.	required

Returns:

Type	Description
NDArray	An array of N bin centers.

Source code in earthcarekit/utils/numpy/_bin.py

def bins_to_centers(bins: ArrayLike) -> NDArray:
    """Converts bin edges to bin centers.

    Args:
        bins: Array of N+1 bin edges.

    Returns:
        An array of N bin centers.
    """
    bins = np.asarray(bins)
    return bins[0:-1] + np.diff(bins) * 0.5

centers_to_bins

centers_to_bins(centers: ArrayLike) -> NDArray

Estimates bin edges from bin centers, assuming edges lie halfway between centers.

Parameters:

Name	Type	Description	Default
centers	ArrayLike	Array of N bin centers.	required

Returns:

Type	Description
NDArray	An array of N+1 bin edges.

Source code in earthcarekit/utils/numpy/_bin.py

def centers_to_bins(centers: ArrayLike) -> NDArray:
    """Estimates bin edges from bin centers, assuming edges lie halfway between centers.

    Args:
        centers: Array of N bin centers.

    Returns:
        An array of N+1 bin edges.
    """
    centers = np.asarray(centers)

    d1 = np.diff(centers)
    d2 = np.append(d1[0], d1)
    d3 = np.append(d1, d1[-1])

    bins1 = centers - d2 / 2
    bins2 = centers + d3 / 2

    return np.append(bins1, bins2[-1])

circular_nanmean

circular_nanmean(degrees: ArrayLike, axis: int | None = None) -> float

Compute the circular mean of angles in degrees, ignoring NaNs.

Source code in earthcarekit/utils/numpy/_circular.py

def circular_nanmean(degrees: ArrayLike, axis: int | None = None) -> float:
    """Compute the circular mean of angles in degrees, ignoring NaNs."""
    radians = np.deg2rad(degrees)
    sin_sum = np.nanmean(np.sin(radians), axis=axis)
    cos_sum = np.nanmean(np.cos(radians), axis=axis)
    mean_angle = np.arctan2(sin_sum, cos_sum)
    return np.rad2deg(mean_angle)

clamp

clamp(a: Scalar, min: Scalar, max: Scalar) -> Scalar

clamp(a: Sequence[Scalar] | NDArray, min: Scalar, max: Scalar) -> NDArray[generic]

clamp(a, min, max)

Limits given values to a range between a minimum and maximum value.

Parameters:

Name	Type	Description	Default
a		Input array or array-like object to be clamped.	required
min		Minimum limit.	required
max		Maximum limit.	required

Returns:

Type	Description
	A clampled scalar or array.

Examples:

>>> import numpy as np
>>> import earthcarekit.utils.numpy as np_utils
>>> x1 = np_utils.clamp(1.2, 0, 1)
>>> print(x1)
1.0

>>> x2 = np_utils.clamp([1.0, 1.2, 0.8, -0.1], 0, 1)
>>> print(x2)
[1.  1.  0.8 0. ]

>>> x3 = np_utils.clamp(
...     np.datetime64("2025-02-23"),
...     np.datetime64("2025-01-01"),
...     np.datetime64("2025-01-02"),
... )
>>> print(x3)
2025-01-02

>>> x4 = np_utils.clamp(
...     np.timedelta64(70, "m"), np.timedelta64(0, "h"), np.timedelta64(1, "h")
... )
>>> print(x4)
60 minutes

Source code in earthcarekit/utils/numpy/_clamp.py

def clamp(a, min, max):
    """Limits given values to a range between a minimum and maximum value.

    Args:
        a: Input array or array-like object to be clamped.
        min: Minimum limit.
        max: Maximum limit.

    Returns:
        A clampled scalar or array.

    Examples:
        >>> import numpy as np
        >>> import earthcarekit.utils.numpy as np_utils
        >>> x1 = np_utils.clamp(1.2, 0, 1)
        >>> print(x1)
        1.0

        >>> x2 = np_utils.clamp([1.0, 1.2, 0.8, -0.1], 0, 1)
        >>> print(x2)
        [1.  1.  0.8 0. ]

        >>> x3 = np_utils.clamp(
        ...     np.datetime64("2025-02-23"),
        ...     np.datetime64("2025-01-01"),
        ...     np.datetime64("2025-01-02"),
        ... )
        >>> print(x3)
        2025-01-02

        >>> x4 = np_utils.clamp(
        ...     np.timedelta64(70, "m"), np.timedelta64(0, "h"), np.timedelta64(1, "h")
        ... )
        >>> print(x4)
        60 minutes
    """
    a = np.asarray(a)
    if np.isnan(max):
        max = np.nanmax(a)
    if np.isnan(min):
        min = np.nanmin(a)
    return np.maximum(np.minimum(np.asarray(a), max), min)

coarsen_mean

coarsen_mean(a: ArrayLike, n: int, axis: int = 0, is_bin: bool = False) -> NDArray

Downsample an array by averaging or taking the mode over n adjacent elements.

Parameters:

Name	Type	Description	Default
a	ArrayLike	Input array or array-like object.	required
n	int	Number of elements per bin.	required
axis	int	Axis along which to downsample.	0
is_bin	bool	If True, treat a as integer classification data and compute the mode (most frequent value) per bin; otherwise, compute the mean.	False

Returns:

Type	Description
NDArray	The downsampled array.

Source code in earthcarekit/utils/numpy/_coarsen_mean.py

def coarsen_mean(
    a: ArrayLike,
    n: int,
    axis: int = 0,
    is_bin: bool = False,
) -> NDArray:
    """Downsample an array by averaging or taking the mode over `n` adjacent elements.

    Args:
        a: Input array or array-like object.
        n: Number of elements per bin.
        axis: Axis along which to downsample.
        is_bin: If True, treat `a` as integer classification data and compute the mode (most frequent value) per bin; otherwise, compute the mean.

    Returns:
        The downsampled array.
    """
    a = np.asarray(a)
    a = np.moveaxis(a, axis, 0)

    # Discard residual data points
    trimmed_len = (a.shape[0] // n) * n
    trimmed = a[:trimmed_len]
    reshaped = trimmed.reshape(-1, n, *a.shape[1:])

    # Average
    is_datetime = np.issubdtype(a.dtype, np.datetime64)
    averaged: NDArray
    if is_datetime:
        averaged = reshaped.astype("datetime64[ns]").astype("int64").mean(axis=1)
        averaged = averaged.astype("datetime64[ns]")
    elif is_bin:
        if len(a.shape) == 1:
            averaged = np.apply_along_axis(get_most_freq_int, 0, reshaped)
        elif len(a.shape) == 2:
            averaged = np.array([np.apply_along_axis(get_most_freq_int, 0, x) for x in reshaped])
    else:
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", category=RuntimeWarning)
            averaged = np.nanmean(reshaped, axis=1)

    return np.moveaxis(averaged, 0, axis)

downsample

downsample(a: ArrayLike, n: int, axis: int = 0) -> NDArray

Downsample an array by selecting evenly spaced samples along one axis.

Parameters:

Name	Type	Description	Default
a	ArrayLike	Input array or array-like object to downsample.	required
n	int	Number of samples to select.	required
axis	int	The axis along which the array a will be downsampled.	0

Returns:

Type	Description
NDArray	The downsampled array.

Source code in earthcarekit/utils/numpy/_downsample.py

def downsample(
    a: ArrayLike,
    n: int,
    axis: int = 0,
) -> NDArray:
    """Downsample an array by selecting evenly spaced samples along one axis.

    Args:
        a: Input array or array-like object to downsample.
        n: Number of samples to select.
        axis: The axis along which the array `a` will be downsampled.

    Returns:
        The downsampled array.
    """
    if n < 1:
        raise ValueError("'n' must be at least 1")

    a = np.asarray(a)
    n_original = a.shape[axis]

    if n_original == 0:
        raise ValueError(f"Can't downsample empty axis ({axis}): {a.shape=})")

    indices = np.rint(np.linspace(0, n_original - 1, n)).astype(np.intp)

    shape = [1] * a.ndim
    shape[axis] = indices.size
    indices = np.reshape(indices, shape)

    return np.take_along_axis(a, indices=indices, axis=axis)

flatten_array

flatten_array(a: ArrayLike) -> NDArray

Flatten a nested sequence of array-likes into a 1D numpy.array.

Parameters:

Name	Type	Description	Default
a	ArrayLike	Sequence of array-like objects (may contain lists, tuples, arrays, or non-iterable elements).	required

Returns:

Type	Description
NDArray	The flattened 1D array.

Source code in earthcarekit/utils/numpy/_flatten.py

def flatten_array(a: ArrayLike) -> NDArray:
    """Flatten a nested sequence of array-likes into a 1D numpy.array.

    Args:
        a: Sequence of array-like objects (may contain lists, tuples, arrays, or non-iterable elements).

    Returns:
        The flattened 1D array.
    """
    if isinstance(a, np.ndarray):
        return a.flatten()

    flattened_sequence = []

    def _ensure_list(x):
        if isinstance(x, list):
            return x.copy()
        return [x].copy()

    stack = _ensure_list(a)  # type: ignore

    while stack:
        item = stack.pop(0)
        if isinstance(item, (list, tuple, np.ndarray)):
            stack = list(item) + stack
        else:
            flattened_sequence.append(item)

    return np.array(flattened_sequence)

get_number_range

get_number_range(
    start: float, end: float, freq: float | None = None, periods: int | None = None
) -> NDArray[floating | integer]

Generates a sequence of numbers based on frequency or number of periods.

Parameters:

Name	Type	Description	Default
start	float	First value in the sequence.	required
end	float	Last value in the sequence.	required
freq	float | None	Sampling frequency.	None
periods	int | None	Number of evenly spaced values.	None

Returns:

Type	Description
NDArray[floating | integer]	A sequence of numbers, either frequency-sampled or evenly spaced.

Source code in earthcarekit/utils/numpy/_misc.py

def get_number_range(
    start: float, end: float, freq: float | None = None, periods: int | None = None
) -> NDArray[np.floating | np.integer]:
    """Generates a sequence of numbers based on frequency or number of periods.

    Args:
        start: First value in the sequence.
        end: Last value in the sequence.
        freq: Sampling frequency.
        periods: Number of evenly spaced values.

    Returns:
        A sequence of numbers, either frequency-sampled or evenly spaced.
    """
    if freq is None and periods is None:
        raise TypeError(
            f"{get_number_range.__name__}() missing 1 required argument: 'freq' or 'periods'"
        )
    elif isinstance(freq, float) and isinstance(periods, int):
        raise TypeError(
            f"{get_number_range.__name__}() expected 1 out of the 2 required arguments 'freq' and 'periods' but got both"
        )
    elif isinstance(freq, float):
        mod = start % freq
        s = start - mod
        if mod != 0.0:
            s += freq
        mod = end % freq
        e = end - mod
        result = np.arange(s, e + freq, freq)
        return np.array(result)
    elif isinstance(periods, int):
        result = np.linspace(start, end, periods)
        return np.array(result)
    else:
        raise RuntimeError(f"{get_number_range.__name__}() implementation error")

isascending

isascending(a: ArrayLike, raise_error: bool = False, result_constant: bool = True) -> bool

Checks whether a sequence is initially ascending.

Parameters:

Name	Type	Description	Default
a	ArrayLike	Input sequence (e.g., list, array).	required
raise_error	bool	Raise ValueError if sequence length < 2.	False
result_constant	bool	Treat constant sequences as ascending if True.	True

Returns:

Type	Description
bool	True if the sequence is ascending, False otherwise.

Raises:

Type	Description
ValueError	If raise_error=True and sequence length < 2.

Source code in earthcarekit/utils/numpy/_check.py

def isascending(
    a: ArrayLike,
    raise_error: bool = False,
    result_constant: bool = True,
) -> bool:
    """Checks whether a sequence is initially ascending.

    Args:
        a: Input sequence (e.g., list, array).
        raise_error: Raise ValueError if sequence length < 2.
        result_constant: Treat constant sequences as ascending if True.

    Returns:
        True if the sequence is ascending, False otherwise.

    Raises:
        ValueError: If `raise_error=True` and sequence length < 2.
    """
    _a: NDArray = np.array(a)
    _a = _a[~np.isnan(_a)]

    if len(_a) < 2:
        if raise_error:
            raise ValueError(f"too few latitude values ({len(_a)}) but at least 2 are needed.")
        return False
    diff = np.diff(_a)
    for d in diff:
        if d > 0:
            return True
        elif d < 0:
            return False
    return result_constant

ismonotonic

ismonotonic(
    a: ArrayLike,
    strict: bool = False,
    mode: Literal["any", "increasing", "decreasing"] = "any",
    raise_error: bool = False,
    ignore_nans: bool = True,
)

Checks whether a sequence is monotonic.

Parameters:

Name	Type	Description	Default
a	ArrayLike	Input sequence (e.g., list, array).	required
strict	bool	Require strictly increasing/decreasing if True.	False
mode	Literal['any', 'increasing', 'decreasing']	Direction to check ("any", "increasing", "decreasing").	'any'
raise_error	bool	Raise ValueError if not monotonic.	False
ignore_nans	bool	Skip NaN values if True.	True

Returns:

Type	Description
	True if monotonic according to parameters, False otherwise.

Raises:

Type	Description
ValueError	If mode is invalid or raise_error=True and sequence is not monotonic.

Source code in earthcarekit/utils/numpy/_check.py

def ismonotonic(
    a: ArrayLike,
    strict: bool = False,
    mode: Literal["any", "increasing", "decreasing"] = "any",
    raise_error: bool = False,
    ignore_nans: bool = True,
):
    """Checks whether a sequence is monotonic.

    Args:
        a: Input sequence (e.g., list, array).
        strict: Require strictly increasing/decreasing if True.
        mode: Direction to check ("any", "increasing", "decreasing").
        raise_error: Raise ValueError if not monotonic.
        ignore_nans: Skip NaN values if True.

    Returns:
        True if monotonic according to parameters, False otherwise.

    Raises:
        ValueError: If `mode` is invalid or `raise_error=True` and sequence is not monotonic.
    """
    a = np.asarray(a)
    if ignore_nans:
        a = a[~np.isnan(a)]

    signs = np.sign(np.diff(a))

    correct_signs = []

    if not strict:
        correct_signs.append(0)

    if mode == "any":
        i: int = 0
        while i < len(signs) - 1 and signs[i] == 0:
            i = i + 1

        if signs[i] != 0:
            correct_signs.append(signs[i])
    elif mode == "increasing":
        correct_signs.append(1)
    elif mode == "decreasing":
        correct_signs.append(-1)
    else:
        raise ValueError(
            f"invalid `mode` ('{mode}') given, but expecting 'any', 'increasing' or 'decreasing'"
        )

    is_monotonic = all([s in correct_signs for s in signs])

    if raise_error and not is_monotonic:
        raise TypeError(
            f"sequence must be monotonic but it is not (strict={strict}, mode='{mode}')"
        )

    return is_monotonic

isndarray

isndarray(a: Any, dtype: DTypeLike | None = None, raise_error: bool = False)

Returns True if a has type numpy.ndarray and also checks if dtype is lower/equal in type hierarchy if given (i.e. returns True if a.dtype is subtype of dtype).

Source code in earthcarekit/utils/numpy/_check.py

def isndarray(a: Any, dtype: DTypeLike | None = None, raise_error: bool = False):
    """
    Returns True if `a` has type `numpy.ndarray` and also checks if `dtype` is lower/equal
    in type hierarchy if given (i.e. returns True if `a.dtype` is subtype of `dtype`).
    """
    if dtype is None:
        is_ndarray = isinstance(a, np.ndarray)
    else:
        is_ndarray = isinstance(a, np.ndarray) and np.issubdtype(a.dtype, dtype)

    if raise_error and not is_ndarray:
        dtype_str = "Any" if dtype is None else str(dtype)
        raise TypeError(
            f"expected type ndarray[{dtype_str}] for `a` but got {type(a).__name__}[{type(a[0]).__name__}]"
        )

    return is_ndarray

lookup_value_by_number

lookup_value_by_number(n: float, numbers: ArrayLike, values: ArrayLike) -> Any

Returns the value corresponding to the closest number in a monotonic series.

Parameters:

Name	Type	Description	Default
n	float	Number to look up.	required
numbers	ArrayLike	Monotonically increasing series of numbers.	required
values	ArrayLike	Values corresponding to numbers.	required

Returns:

Type	Description
Any	Value from values at the index of the closest number in numbers.

Raises:

Type	Description
ValueError	If numbers and values have different lengths.

Source code in earthcarekit/utils/numpy/_misc.py

def lookup_value_by_number(n: float, numbers: ArrayLike, values: ArrayLike) -> Any:
    """Returns the value corresponding to the closest number in a monotonic series.

    Args:
        n: Number to look up.
        numbers: Monotonically increasing series of numbers.
        values: Values corresponding to `numbers`.

    Returns:
        Value from `values` at the index of the closest number in `numbers`.

    Raises:
        ValueError: If `numbers` and `values` have different lengths.
    """
    if n is None:
        raise ValueError(f"{lookup_value_by_number.__name__}() missing `n`")
    if numbers is None:
        raise ValueError(f"{lookup_value_by_number.__name__}() missing `numbers`")
    if values is None:
        raise ValueError(f"{lookup_value_by_number.__name__}() missing `values`")

    n = float(n)
    numbers = np.asarray(numbers)
    values = np.asarray(values)

    if numbers.shape[0] == 0:
        raise ValueError(
            f"{lookup_value_by_number.__name__}() `numbers` is empty but needs at least on element"
        )
    if values.shape[0] != numbers.shape[0]:
        raise ValueError(
            f"{lookup_value_by_number.__name__}() First shapes must be the same for `values` ({values.shape[0]}) and `numbers` ({numbers.shape[0]})"
        )

    idx0 = int(np.searchsorted(numbers, n).astype(int) - 1)
    idx1 = int(np.searchsorted(numbers, n).astype(int))

    idx0 = int(np.min([len(numbers) - 1, np.max([0, idx0])]))
    idx1 = int(np.min([len(numbers) - 1, np.max([0, idx1])]))

    if numbers[idx0] > n:
        idx0 = idx1

    total_diff = numbers[idx1] - numbers[idx0]

    diff = n - numbers[idx0]

    if total_diff == 0:
        frac = 0
    else:
        frac = diff / total_diff

    total_amount = values[idx1] - values[idx0]
    v = values[idx0] + total_amount * frac

    return v

normalize

normalize(a: ArrayLike, vmin: float = 0, vmax: float = 1) -> NDArray

Normalizes a sequence to [vmin, vmax], preserving NaNs.

Parameters:

Name	Type	Description	Default
a	ArrayLike	Input sequence (may contain NaNs).	required
vmin	float	Minimum of output range; defaults to 0.	0
vmax	float	Maximum of output range; defaults to 1.	1

Returns:

Type	Description
NDArray	Normalized array with NaNs preserved.

Raises:

Type	Description
ValueError	If vmin >= vmax.

Source code in earthcarekit/utils/numpy/_normalize.py

def normalize(
    a: ArrayLike,
    vmin: float = 0,
    vmax: float = 1,
) -> NDArray:
    """Normalizes a sequence to [vmin, vmax], preserving NaNs.

    Args:
        a: Input sequence (may contain NaNs).
        vmin: Minimum of output range; defaults to 0.
        vmax: Maximum of output range; defaults to 1.

    Returns:
        Normalized array with NaNs preserved.

    Raises:
        ValueError: If `vmin >= vmax`.
    """
    if vmin >= vmax:
        raise ValueError(f"vmin ({vmin}) must be smaller than vmax ({vmax})")

    a_old = np.asarray(a, dtype=float)
    vmin_old = np.nanmin(a_old)
    vmax_old = np.nanmax(a_old)

    if np.isnan(vmin_old) or vmin_old == vmax_old:
        a_new = np.full_like(a_old, np.nan)
    else:
        a_new = (a_old - vmin_old) / (vmax_old - vmin_old)

    # Scale
    a_new = a_new * (vmax - vmin)

    # Shift
    a_new = a_new + vmin

    return a_new

pad_true_sequence

pad_true_sequence(a: NDArray[bool_], n: int) -> NDArray[bool_]

Pads all sequences of True values occuring in an array with n True values before and after the sequence (while keeping the original size).

Source code in earthcarekit/utils/numpy/_true_sequence.py

def pad_true_sequence(
    a: NDArray[np.bool_],
    n: int,
) -> NDArray[np.bool_]:
    """Pads all sequences of True values occuring in an array with n True values before and after the sequence (while keeping the original size)."""
    idx = np.flatnonzero(a)
    if idx.size == 0:
        return a.copy()
    mask: NDArray[np.bool_] = np.full(a.shape, False, dtype=bool)
    start = max(0, idx[0] - n)
    end = min(len(a), idx[-1] + n + 1)
    mask[start:end] = True
    return mask

pad_true_sequence_2d

pad_true_sequence_2d(a: NDArray[bool_], n: int, axis: int = 1) -> NDArray[bool_]

Pads all sequences of True values occuring along one axis in an 2d array with n True values before and after the sequence (while keeping the original size).

Source code in earthcarekit/utils/numpy/_true_sequence.py

def pad_true_sequence_2d(
    a: NDArray[np.bool_],
    n: int,
    axis: int = 1,
) -> NDArray[np.bool_]:
    """Pads all sequences of True values occuring along one axis in an 2d array with n True values before and after the sequence (while keeping the original size)."""

    if axis not in [0, 1]:
        raise ValueError(f"invalid axis ({axis}), expected 0 or 1")

    if axis == 0:
        a = a.T

    ncols = a.shape[1]
    idx = np.arange(ncols)

    any_true = np.asarray(a.any(axis=1))

    first_true = np.argmax(a, axis=1)
    first_true = np.where(any_true, first_true, ncols)

    last_true = ncols - 1 - np.argmax(a[:, ::-1], axis=1)
    last_true = np.where(any_true, last_true, -1)

    start = np.clip(first_true - n, 0, ncols)
    end = np.clip(last_true + n + 1, 0, ncols)

    mask = (idx >= start[:, np.newaxis]) & (idx < end[:, np.newaxis])
    mask &= any_true[:, np.newaxis]

    if axis == 0:
        mask = mask.T

    return mask

rebin_lerp

rebin_lerp(
    v: ArrayLike,
    axis0_coords: ArrayLike,
    rebin_index: ArrayLike | None = None,
    bin_edges: ArrayLike | None = None,
    bin_centers: ArrayLike | None = None,
) -> NDArray

Rebin 1D or 2D arrays along the first axis (0) by linearly interpolating the two samples around a bin center.

Parameters:

Name	Type	Description	Default
v	ArrayLike	1D or 2D array to be rebinned.	required
axis0_coords	ArrayLike	Array of reference monotonic values.	required
rebin_index	ArrayLike | None	Array of non-decreasing indecies mapping values in v to target bins. Defaults to None.	None
bin_edges	ArrayLike | None	Array of N+1 bin edges. Defaults to None.	None
bin_centers	ArrayLike | None	Array of N bin centers. Defaults to None.	None

Returns:

Type	Description
NDArray	Along axis 0 rebinned version of original array v.

Source code in earthcarekit/utils/numpy/_rebin/_rebin_lerp.py

def rebin_lerp(
    v: ArrayLike,
    axis0_coords: ArrayLike,
    rebin_index: ArrayLike | None = None,
    bin_edges: ArrayLike | None = None,
    bin_centers: ArrayLike | None = None,
) -> NDArray:
    """Rebin 1D or 2D arrays along the first axis (0) by linearly interpolating the two samples around a bin center.

    Args:
        v: 1D or 2D array to be rebinned.
        axis0_coords: Array of reference monotonic values.
        rebin_index: Array of non-decreasing indecies mapping values in `v` to target bins. Defaults to None.
        bin_edges: Array of N+1 bin edges. Defaults to None.
        bin_centers: Array of N bin centers. Defaults to None.

    Returns:
        Along axis 0 rebinned version of original array `v`.
    """
    v = np.asarray(v)
    axis0_coords = np.asarray(axis0_coords)

    if bin_centers is not None:
        bin_centers = np.asarray(bin_centers)
    elif bin_edges is not None:
        bin_centers = bins_to_centers(bin_edges)
    elif rebin_index is not None:
        rebin_index = np.asarray(rebin_index)
        bin_centers = bins_to_centers(
            axis0_coords[np.append(np.unique(rebin_index, return_index=True)[1], len(rebin_index))]
        )
    else:
        raise ValueError("requires either 'bin_centers', 'bin_edges' or 'rebin_index'")

    idx, w = _get_lerp_params(axis0_coords, bin_centers)
    if v.ndim == 2:
        return _rebin_lerp_2d(idx, w, v)
    elif v.ndim == 1:
        return _rebin_lerp_1d(idx, w, v)
    else:
        raise ValueError(f"{v.ndim}d array not supported: requires 1d or 2d")

rebin_mean

rebin_mean(
    v: ArrayLike,
    rebin_index: ArrayLike | None = None,
    axis0_coords: ArrayLike | None = None,
    bin_edges: ArrayLike | None = None,
    bin_centers: ArrayLike | None = None,
    ignore_nans: bool = True,
) -> NDArray

Rebins 1D or 2D arrays along axis 0 by averaging samples within bins.

Parameters:

Name	Type	Description	Default
v	ArrayLike	1D or 2D array to rebin.	required
rebin_index	ArrayLike | None	Bin indices mapping v to target bins; derived if None.	None
axis0_coords	ArrayLike | None	Reference values for deriving rebin_index; required if rebin_index is None.	None
bin_edges	ArrayLike | None	Bin edges (N+1); used to derive rebin_index if given.	None
bin_centers	ArrayLike | None	Bin centers (N); used to derive rebin_index if given.	None
ignore_nans	bool	Ignore NaNs during averaging if True; bins with NaN return NaN otherwise.	True

Returns:

Type	Description
NDArray	Rebinned array along axis 0.

Source code in earthcarekit/utils/numpy/_rebin/_rebin_mean.py

def rebin_mean(
    v: ArrayLike,
    rebin_index: ArrayLike | None = None,
    axis0_coords: ArrayLike | None = None,
    bin_edges: ArrayLike | None = None,
    bin_centers: ArrayLike | None = None,
    ignore_nans: bool = True,
) -> NDArray:
    """Rebins 1D or 2D arrays along axis 0 by averaging samples within bins.

    Args:
        v: 1D or 2D array to rebin.
        rebin_index: Bin indices mapping `v` to target bins; derived if None.
        axis0_coords: Reference values for deriving `rebin_index`; required if `rebin_index` is None.
        bin_edges: Bin edges (N+1); used to derive `rebin_index` if given.
        bin_centers: Bin centers (N); used to derive `rebin_index` if given.
        ignore_nans: Ignore NaNs during averaging if True; bins with NaN return NaN otherwise.

    Returns:
        Rebinned array along axis 0.
    """
    return _rebin(
        func1d=_rebin_mean_1d,
        func2d=_rebin_mean_2d,
        v=v,
        rebin_index=rebin_index,
        axis0_coords=axis0_coords,
        bin_edges=bin_edges,
        bin_centers=bin_centers,
        ignore_nans=ignore_nans,
    )

rebin_median

rebin_median(
    v: ArrayLike,
    rebin_index: ArrayLike | None = None,
    axis0_coords: ArrayLike | None = None,
    bin_edges: ArrayLike | None = None,
    bin_centers: ArrayLike | None = None,
    ignore_nans: bool = True,
) -> NDArray

Rebins 1D or 2D arrays along axis 0 by computing the median within bins.

Parameters:

Name	Type	Description	Default
v	ArrayLike	1D or 2D array to rebin.	required
rebin_index	ArrayLike | None	Bin indices mapping v to target bins; derived if None.	None
axis0_coords	ArrayLike | None	Reference values for deriving rebin_index; required if rebin_index is None.	None
bin_edges	ArrayLike | None	Bin edges (N+1); used to derive rebin_index if given.	None
bin_centers	ArrayLike | None	Bin centers (N); used to derive rebin_index if given.	None
ignore_nans	bool	Ignore NaNs during median search if True; bins with NaN return NaN otherwise.	True

Returns:

Type	Description
NDArray	Rebinned array along axis 0.

Source code in earthcarekit/utils/numpy/_rebin/_rebin_median.py

def rebin_median(
    v: ArrayLike,
    rebin_index: ArrayLike | None = None,
    axis0_coords: ArrayLike | None = None,
    bin_edges: ArrayLike | None = None,
    bin_centers: ArrayLike | None = None,
    ignore_nans: bool = True,
) -> NDArray:
    """Rebins 1D or 2D arrays along axis 0 by computing the median within bins.

    Args:
        v: 1D or 2D array to rebin.
        rebin_index: Bin indices mapping `v` to target bins; derived if None.
        axis0_coords: Reference values for deriving `rebin_index`; required if `rebin_index` is None.
        bin_edges: Bin edges (N+1); used to derive `rebin_index` if given.
        bin_centers: Bin centers (N); used to derive `rebin_index` if given.
        ignore_nans: Ignore NaNs during median search if True; bins with NaN return NaN otherwise.

    Returns:
        Rebinned array along axis 0.
    """
    return _rebin(
        func1d=_rebin_median_1d,
        func2d=_rebin_median_2d,
        v=v,
        rebin_index=rebin_index,
        axis0_coords=axis0_coords,
        bin_edges=bin_edges,
        bin_centers=bin_centers,
        ignore_nans=ignore_nans,
    )

rescale

rescale(
    a: Scalar1, src_min: Scalar1, src_max: Scalar1, dst_min: Scalar2, dst_max: Scalar2
) -> Scalar2

rescale(
    a: Sequence[Scalar1] | NDArray,
    src_min: Scalar1,
    src_max: Scalar1,
    dst_min: Scalar2,
    dst_max: Scalar2,
) -> NDArray[generic]

rescale(a, src_min, src_max, dst_min, dst_max)

Linearly map values from one range to another.

Parameters:

Name	Type	Description	Default
a		Scalar or array-like values to be rescaled.	required
src_min		Lower bound of the source range.	required
src_max		Upper bound of the source range.	required
dst_min		Lower bound of the destination range.	required
dst_max		Upper bound of the destination range.	required

Returns:

Type	Description
	The rescaled scalar or array. The retrun type matches the type of the destination range.

Examples:

>>> import numpy as np
>>> import earthcarekit.utils.numpy as np_utils
>>> x1 = np_utils.rescale(3, 0, 10, 0, 200)
>>> print(x1)
60.0

>>> x2 = np_utils.rescale([3], 0, 10, 0, 200)
>>> print(x2)
[60.]

>>> x3 = np_utils.rescale(
...     [0, 0.5, 1],
...     0,
...     1,
...     np.datetime64("2025-01-01 00:00:00"),
...     np.datetime64("2025-01-02 00:00:00"),
... )
>>> print(x3)
['2025-01-01T00:00:00' '2025-01-01T12:00:00' '2025-01-02T00:00:00']

>>> x4 = np_utils.rescale(
...     np.timedelta64(15, "m"),
...     np.timedelta64(0, "h"),
...     np.timedelta64(1, "h"),
...     0,
...     1,
... )
>>> print(x4)
0.25

Source code in earthcarekit/utils/numpy/_rescale.py

def rescale(a, src_min, src_max, dst_min, dst_max):
    """Linearly map values from one range to another.

    Args:
        a: Scalar or array-like values to be rescaled.
        src_min: Lower bound of the source range.
        src_max: Upper bound of the source range.
        dst_min: Lower bound of the destination range.
        dst_max: Upper bound of the destination range.

    Returns:
        The rescaled scalar or array. The retrun type matches the type of the destination range.

    Examples:
        >>> import numpy as np
        >>> import earthcarekit.utils.numpy as np_utils
        >>> x1 = np_utils.rescale(3, 0, 10, 0, 200)
        >>> print(x1)
        60.0

        >>> x2 = np_utils.rescale([3], 0, 10, 0, 200)
        >>> print(x2)
        [60.]

        >>> x3 = np_utils.rescale(
        ...     [0, 0.5, 1],
        ...     0,
        ...     1,
        ...     np.datetime64("2025-01-01 00:00:00"),
        ...     np.datetime64("2025-01-02 00:00:00"),
        ... )
        >>> print(x3)
        ['2025-01-01T00:00:00' '2025-01-01T12:00:00' '2025-01-02T00:00:00']

        >>> x4 = np_utils.rescale(
        ...     np.timedelta64(15, "m"),
        ...     np.timedelta64(0, "h"),
        ...     np.timedelta64(1, "h"),
        ...     0,
        ...     1,
        ... )
        >>> print(x4)
        0.25
    """
    a = np.asarray(a)
    with np.errstate(invalid="ignore"):
        fraction = (a - src_min) / (src_max - src_min)
    return dst_min + fraction * (dst_max - dst_min)

shift_true_sequence

shift_true_sequence(a: NDArray[bool_], n: int) -> NDArray[bool_]

Offsets all sequences of True values occuring in an array (while keeping the original size).

Source code in earthcarekit/utils/numpy/_true_sequence.py

def shift_true_sequence(
    a: NDArray[np.bool_],
    n: int,
) -> NDArray[np.bool_]:
    """Offsets all sequences of True values occuring in an array (while keeping the original size)."""
    if n == 0:
        return a

    a = np.roll(a, n)
    if n > 0:
        a[:n] = False
    elif n < 0:
        a[n:] = False

    return a

wrap_to_interval

wrap_to_interval(a: ArrayLike, min: float, max: float) -> NDArray

Wrap values in a to the interval [min, max).

Source code in earthcarekit/utils/numpy/_circular.py

def wrap_to_interval(a: ArrayLike, min: float, max: float) -> NDArray:
    """Wrap values in `a` to the interval [`min`, `max`)."""
    a = np.array(a)
    interval = max - min
    return (a - min) % interval + min