State Diagnostics

state_diagnostics

Opt-in checks of states at requested output times.

These diagnostics do not inspect internal RK stages, rejected attempts or unobserved substeps, and do not project or clamp the numerical solution.

OutputStateCheckConfig(nonnegative_indices=None, monotone_indices=(), negative_tolerance=0.0, monotone_tolerance=0.0) dataclass

Components and tolerances for checks at requested output times.

Attributes:

Name	Type	Description
nonnegative_indices	tuple[int, ...] | None	State components expected to be nonnegative; None selects every component. Selectors address the state axis.
monotone_indices	tuple[int, ...]	Components expected to be nondecreasing, such as cumulative counters. Empty disables increment checks.
negative_tolerance	float	A value below minus this absolute tolerance flags an output. This changes the diagnostic threshold, not the state.
monotone_tolerance	float	An increment below minus this absolute tolerance flags an output interval. It is not scaled by interval duration.

__post_init__()

Detach selectors and validate finite nonnegative tolerances.

Raises:

Type	Description
TypeError	If a selector or tolerance has the wrong type.
ValueError	If a selector or tolerance is invalid.

Source code in src/op_engine/state_diagnostics.py

def __post_init__(self) -> None:
    """Detach selectors and validate finite nonnegative tolerances.

    Raises:
        TypeError: If a selector or tolerance has the wrong type.
        ValueError: If a selector or tolerance is invalid.
    """
    if self.nonnegative_indices is not None:
        object.__setattr__(
            self,
            "nonnegative_indices",
            _indices(
                self.nonnegative_indices,
                name="nonnegative_indices",
                allow_empty=False,
            ),
        )
    object.__setattr__(
        self,
        "monotone_indices",
        _indices(self.monotone_indices, name="monotone_indices", allow_empty=True),
    )
    for name in ("negative_tolerance", "monotone_tolerance"):
        value = getattr(self, name)
        if isinstance(value, bool) or not isinstance(value, Real):
            msg = f"{name} must be a finite nonnegative number"
            raise TypeError(msg)
        normalized = float(value)
        if not np.isfinite(normalized) or normalized < 0:
            msg = f"{name} must be a finite nonnegative number"
            raise ValueError(msg)
        object.__setattr__(self, name, normalized)

OutputStateDiagnostics

Bases: NamedTuple

Per-subgroup/field-cell checks over the requested output grid.

Arrays retain every state axis except the state-component axis. Counts count outputs or intervals with at least one flagged component, not the number of components. Values and flags stay in the input namespace.

Attributes:

Name	Type	Description
n_outputs	int	Number of inspected outputs, including the initial state.
complete	bool	Whether every requested output was inspected.
min_state	Array	Minimum selected nonnegative-state value across outputs. NaNs propagate; inspect nonfinite counts too.
n_negative_outputs	Array	Outputs with any selected state below tolerance.
n_nonfinite_outputs	Array	Outputs with any nonfinite state component, including components outside nonnegative_indices.
min_increment	Array | None	Minimum increment of the selected monotone components, or None if disabled. Positive infinity when no interval exists.
n_decreasing_intervals	Array | None	Output intervals with any selected decreasing component, or None if monotonicity checks are disabled.

valid property

Return native flags for complete, finite, unflagged output records.

diagnose_trajectory(trajectory, config=None)

Check an output trajectory without modifying it or extracting host scalars.

Parameters:

Name	Type	Description	Default
trajectory	Array	Array shaped (output, state, *subgroups_or_field_axes).	required
config	OutputStateCheckConfig | None	Component selectors and absolute tolerances.	None

Returns:

Type	Description
OutputStateDiagnostics	Diagnostics for the complete supplied trajectory. valid describes
OutputStateDiagnostics	these recorded outputs, not intermediate-step positivity or accuracy.

Raises:

Type	Description
TypeError	If config has the wrong type.
ValueError	If the trajectory shape or selectors are invalid.

Examples:

>>> import numpy as np
>>> path = np.array([[[1.0, 1.0]], [[0.9, -1.0]], [[0.81, 1.0]]])
>>> result = diagnose_trajectory(path)
>>> result.n_negative_outputs.tolist()
[0, 1]
>>> result.valid.tolist()
[True, False]

Source code in src/op_engine/state_diagnostics.py

def diagnose_trajectory(
    trajectory: Array,
    config: OutputStateCheckConfig | None = None,
) -> OutputStateDiagnostics:
    """Check an output trajectory without modifying it or extracting host scalars.

    Args:
        trajectory: Array shaped ``(output, state, *subgroups_or_field_axes)``.
        config: Component selectors and absolute tolerances.

    Returns:
        Diagnostics for the complete supplied trajectory. ``valid`` describes
        these recorded outputs, not intermediate-step positivity or accuracy.

    Raises:
        TypeError: If config has the wrong type.
        ValueError: If the trajectory shape or selectors are invalid.

    Examples:
        >>> import numpy as np
        >>> path = np.array([[[1.0, 1.0]], [[0.9, -1.0]], [[0.81, 1.0]]])
        >>> result = diagnose_trajectory(path)
        >>> result.n_negative_outputs.tolist()
        [0, 1]
        >>> result.valid.tolist()
        [True, False]
    """
    cfg = OutputStateCheckConfig() if config is None else config
    if not isinstance(cfg, OutputStateCheckConfig):
        msg = "config must be an OutputStateCheckConfig"
        raise TypeError(msg)
    if len(trajectory.shape) < 2 or trajectory.shape[0] < 1:
        msg = "Trajectory must have nonempty output and state axes"
        raise ValueError(msg)
    _validate_components(cfg, trajectory.shape[1])
    xp = array_namespace(trajectory)
    if not xp.isdtype(trajectory.dtype, "real floating"):
        msg = "Output-state diagnostics require real floating states"
        raise TypeError(msg)
    values = xp.asarray(trajectory)
    selected = (
        values
        if cfg.nonnegative_indices is None
        else values[:, xp.asarray(cfg.nonnegative_indices, dtype=xp.int32)]
    )
    minimum = xp.min(xp.min(selected, axis=1), axis=0)
    negatives = xp.sum(
        xp.astype(xp.any(selected < -cfg.negative_tolerance, axis=1), xp.int32), axis=0
    )
    nonfinite = xp.sum(
        xp.astype(xp.any(~xp.isfinite(values), axis=1), xp.int32), axis=0
    )
    min_increment = decreasing = None
    if cfg.monotone_indices:
        monotone = values[:, xp.asarray(cfg.monotone_indices, dtype=xp.int32)]
        if trajectory.shape[0] == 1:
            min_increment = xp.full_like(minimum, float("inf"))
            decreasing = xp.zeros_like(negatives)
        else:
            increments = monotone[1:] - monotone[:-1]
            min_increment = xp.min(xp.min(increments, axis=1), axis=0)
            decreasing = xp.sum(
                xp.astype(
                    xp.any(increments < -cfg.monotone_tolerance, axis=1), xp.int32
                ),
                axis=0,
            )
    return OutputStateDiagnostics(
        n_outputs=int(trajectory.shape[0]),
        complete=True,
        min_state=minimum,
        n_negative_outputs=negatives,
        n_nonfinite_outputs=nonfinite,
        min_increment=min_increment,
        n_decreasing_intervals=decreasing,
    )