Skip to content

check_monotonic

Classes:

Name Description
CheckMonotonic

Classes#

CheckMonotonic #

Bases: QualityChecker


Checks if any values are not ordered strictly monotonically (i.e. values must all be increasing or all decreasing). The check marks values as failed if they break from a monotonic order.


Classes:

Name Description
Parameters

Methods:

Name Description
get_axis
run

Attributes:

Name Type Description
parameters Parameters

Attributes#

parameters class-attribute instance-attribute #
parameters: Parameters = Parameters()

Classes#

Parameters #

Bases: BaseModel

Methods:

Name Description
check_monotonic_not_increasing_and_decreasing

Attributes:

Name Type Description
dim Optional[str]
require_decreasing bool
require_increasing bool
Attributes#
dim class-attribute instance-attribute #
dim: Optional[str] = None
require_decreasing class-attribute instance-attribute #
require_decreasing: bool = False
require_increasing class-attribute instance-attribute #
require_increasing: bool = False
Functions#
check_monotonic_not_increasing_and_decreasing #
check_monotonic_not_increasing_and_decreasing(
    inc: bool, values: Dict[str, Any]
) -> bool
Source code in tsdat/qc/checkers/check_monotonic.py
@validator("require_increasing")
def check_monotonic_not_increasing_and_decreasing(
    cls, inc: bool, values: Dict[str, Any]
) -> bool:
    if inc and values["require_decreasing"]:
        raise ValueError(
            "CheckMonotonic -> Parameters: cannot set both 'require_increasing'"
            " and 'require_decreasing'. Please set one or both to False."
        )
    return inc

Functions#

get_axis #
get_axis(variable: xr.DataArray) -> int
Source code in tsdat/qc/checkers/check_monotonic.py
def get_axis(self, variable: xr.DataArray) -> int:
    if not (dim := self.parameters.dim):
        dim = variable.dims[0]  # type: ignore
    return variable.get_axis_num(dim)  # type: ignore
run #
run(
    dataset: xr.Dataset, variable_name: str
) -> Union[NDArray[np.bool_], None]
Source code in tsdat/qc/checkers/check_monotonic.py
def run(
    self,
    dataset: xr.Dataset,
    variable_name: str,
) -> Union[NDArray[np.bool_], None]:
    variable = dataset[variable_name]
    failures = np.full(variable.shape, False)

    if self.parameters.dim is None and len(variable.shape) == 2:
        logger.warning(
            "Variable '%s' has shape '%s'. 2D variables must provide a 'dim'"
            " parameter for the name of the dimension to check the variable for"
            " monotonicity.",
            variable_name,
            variable.shape,
        )
        return None

    if variable.values.dtype.kind in {"U", "S"}:  # type: ignore
        logger.warning(
            "Variable '%s' has dtype '%s', which is currently not supported for"
            " monotonicity checks.",
            variable_name,
            variable.values.dtype,  # type: ignore
        )
        return None

    axis = self.get_axis(variable)
    zero = np.timedelta64(0) if is_datetime_like(variable.data) else 0

    # TODO: `direction` would be better assigned as a class var on init than within a method.
    direction: Literal["increasing", "decreasing", ""] = ""

    if self.parameters.require_decreasing:
        direction = "decreasing"
    elif self.parameters.require_increasing:
        direction = "increasing"
    else:
        diff = np.diff(variable.data, axis=axis)  # type: ignore
        direction = (
            "increasing"
            if np.sum(diff > zero) >= np.sum(diff < zero)
            else "decreasing"
        )

    # Find all the values where things break, not just those flagged by diff
    # if any(failures) and not all(failures):
    if len(variable.shape) == 1:
        prev = variable.values[0]
        for i, value in enumerate(variable.values[1:]):
            success = value < prev if direction == "decreasing" else value > prev
            if success:
                prev = value  # only update prev on success
            else:
                failures[i + 1] = True
    else:
        # 2D diff isn't as clever with failing indexes; just report all individual
        # points that fail
        diff = np.gradient(variable.data)[axis]
        failures = diff <= zero if direction == "increasing" else diff >= zero

    return failures

Functions#