Add validity check functions for ContinuousConstraints

CHANGELOG.md

@@ -36,6 +36,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   inverting the specification to keep the complement
 - `DiscreteSelectionConstraint` as the condition-based filtering constraint
   (inclusion-by-default; replaces `DiscreteExcludeConstraint`)
+- `get_invalid`/`get_valid` methods on `ContinuousConstraint` and
+  `ContinuousLinearConstraint`, enabling row-level validation of continuous linear
+  constraints against a dataframe of candidate configurations
 
 ### Changed
 - `BOTORCH` GP preset now includes `BetaPrior(2.5, 1.5)` for the task covariance

baybe/constraints/base.py

@@ -236,6 +236,31 @@ class ContinuousConstraint(Constraint, ABC):
     numerical_only: ClassVar[bool] = True
     # See base class.
 
+    @abstractmethod
+    def get_invalid(self, df: pd.DataFrame, /) -> pd.Index:
+        """Get the indices of dataframe entries that violate the constraint.
+
+        Args:
+            df: A dataframe where each row represents a parameter configuration.
+
+        Raises:
+            ValueError: If the dataframe is missing required parameter columns.
+
+        Returns:
+            The dataframe indices of rows that violate the constraint.
+        """
+
+    def get_valid(self, df: pd.DataFrame, /) -> pd.Index:
+        """Get the indices of dataframe entries that satisfy the constraint.
+
+        Args:
+            df: A dataframe where each row represents a parameter configuration.
+
+        Returns:
+            The dataframe indices of rows that fulfill the constraint.
+        """
+        return df.index.drop(self.get_invalid(df))
+
 
 @define(slots=False)
 class CardinalityConstraint(Constraint, ABC):

baybe/constraints/continuous.py

@@ -11,15 +11,17 @@
 
 import cattrs
 import numpy as np
+import pandas as pd
 from attrs import define, evolve, field
 from attrs.validators import deep_iterable, gt, in_, instance_of, lt
-from typing_extensions import Self
+from typing_extensions import Self, override
 
 from baybe.constraints.base import (
     CardinalityConstraint,
     ContinuousConstraint,
     ContinuousNonlinearConstraint,
 )
+from baybe.constraints.conditions import ThresholdCondition
 from baybe.parameters import NumericalContinuousParameter
 from baybe.settings import active_settings
 from baybe.utils.interval import Interval
@@ -197,6 +199,41 @@ def to_botorch(
             # https://github.com/pytorch/botorch/blob/1518b304f47f5cdbaf9c175e808c90b3a0a6b86d/botorch/optim/optimize.py#L609 # noqa: E501
             return [(idxs_batched_2d, coefficients_batched, rhs)]
 
+    @override
+    def get_invalid(self, df: pd.DataFrame, /) -> pd.Index:
+        """Get the indices of dataframe entries that violate the constraint.
+
+        For intrapoint constraints, each row is evaluated independently.
+        For interpoint constraints, the aggregate across all rows is compared
+        against :attr:`rhs`; if violated, all rows are returned as invalid.
+
+        Args:
+            df: A dataframe where each row represents a parameter configuration.
+
+        Raises:
+            ValueError: If the dataframe is missing required parameter columns.
+
+        Returns:
+            The dataframe indices of rows that violate the constraint.
+        """
+        if missing := self._required_parameters - set(df.columns):
+            raise ValueError(
+                f"'{self.__class__.__name__}' requires columns {missing} "
+                f"which are missing from the dataframe."
+            )
+        series = pd.Series(
+            sum(
+                df[p].to_numpy() * c for p, c in zip(self.parameters, self.coefficients)
+            ),
+            index=df.index,
+        )
+        condition = ThresholdCondition(threshold=self.rhs, operator=self.operator)
+        if self.is_interpoint:
+            # Aggregate across all batch rows (mirrors BoTorch interpoint semantics).
+            satisfied = bool(condition.evaluate(pd.Series([float(series.sum())]))[0])
+            return pd.Index([]) if satisfied else df.index
+        return df.index[~condition.evaluate(series)]
+
 
 @define
 class ContinuousCardinalityConstraint(
@@ -313,6 +350,27 @@ def get_absolute_thresholds(self, bounds: Interval, /) -> Interval:
             upper=self.relative_threshold * bounds.upper,
         )
 
+    @override
+    def get_invalid(self, df: pd.DataFrame, /) -> pd.Index:
+        """Not supported: bounds are required to determine activity thresholds.
+
+        Use :func:`baybe.constraints.utils.is_cardinality_fulfilled` instead.
+
+        Args:
+            df: A dataframe where each row represents a parameter configuration.
+
+        Raises:
+            NotImplementedError: Always.
+
+        Returns:
+            None. Always raises NotImplementedError.
+        """
+        raise NotImplementedError(
+            f"'{self.__class__.__name__}' cannot evaluate row-level validity without "
+            f"parameter bounds. Use 'is_cardinality_fulfilled' from "
+            f"'baybe.constraints.utils' instead."
+        )
+
 
 # Collect leftover original slotted classes processed by `attrs.define`
 gc.collect()

tests/constraints/test_constraints_continuous.py

@@ -1,11 +1,15 @@
 """Test for imposing continuous constraints."""
 
 import numpy as np
+import pandas as pd
 import pytest
 import torch
 from pytest import param
 
-from baybe.constraints import ContinuousLinearConstraint
+from baybe.constraints import (
+    ContinuousCardinalityConstraint,
+    ContinuousLinearConstraint,
+)
 from baybe.parameters.numerical import NumericalContinuousParameter
 from tests.conftest import run_iterations
 
@@ -230,3 +234,112 @@ def test_invalid_constraints(parameters, coefficients, rhs, op):
         ContinuousLinearConstraint(
             parameters=parameters, operator=op, coefficients=coefficients, rhs=rhs
         )
+
+
+@pytest.mark.parametrize(
+    ("operator", "coefficients", "rhs", "df_values", "expected_invalid_indices"),
+    [
+        param(
+            "=",
+            [1.0, 1.0],
+            1.0,
+            {"x": [0.5, 0.3], "y": [0.5, 0.7]},
+            [],
+            id="eq_satisfied",
+        ),
+        param(
+            "=",
+            [1.0, 1.0],
+            1.0,
+            {"x": [0.5, 0.3], "y": [0.5, 0.3]},
+            [1],
+            id="eq_violated",
+        ),
+        param(
+            ">=",
+            [1.0, 1.0],
+            1.0,
+            {"x": [0.5, 0.6], "y": [0.5, 0.6]},
+            [],
+            id="ge_satisfied",
+        ),
+        param(
+            ">=",
+            [1.0, 1.0],
+            1.0,
+            {"x": [0.3, 0.6], "y": [0.3, 0.6]},
+            [0],
+            id="ge_violated",
+        ),
+        param(
+            "<=",
+            [1.0, 1.0],
+            1.0,
+            {"x": [0.4, 0.6], "y": [0.4, 0.6]},
+            [1],
+            id="le_violated",
+        ),
+    ],
+)
+def test_linear_constraint_get_invalid(
+    operator, coefficients, rhs, df_values, expected_invalid_indices
+):
+    """Test get_invalid returns the correct violating row indices."""
+    constraint = ContinuousLinearConstraint(
+        parameters=list(df_values.keys()),
+        operator=operator,
+        coefficients=coefficients,
+        rhs=rhs,
+    )
+    df = pd.DataFrame(df_values)
+    assert list(constraint.get_invalid(df)) == expected_invalid_indices
+
+
+def test_linear_constraint_get_invalid_missing_column():
+    """Test get_invalid raises ValueError for missing required columns."""
+    constraint = ContinuousLinearConstraint(
+        parameters=["x", "y"], operator="=", rhs=1.0
+    )
+    df = pd.DataFrame({"x": [0.5]})
+    with pytest.raises(ValueError, match="missing"):
+        constraint.get_invalid(df)
+
+
+def test_linear_constraint_get_valid_is_complement():
+    """Test get_valid returns the complement of get_invalid."""
+    constraint = ContinuousLinearConstraint(
+        parameters=["x", "y"], operator=">=", rhs=1.0
+    )
+    df = pd.DataFrame({"x": [0.3, 0.6, 0.1], "y": [0.3, 0.6, 0.1]})
+    assert constraint.get_invalid(df).union(constraint.get_valid(df)).equals(df.index)
+
+
+def test_linear_interpoint_constraint_get_invalid_satisfied():
+    """Test interpoint get_invalid returns empty index when aggregate satisfies."""
+    constraint = ContinuousLinearConstraint(
+        parameters=["x"], operator="=", rhs=1.0, interpoint=True
+    )
+    df = pd.DataFrame({"x": [0.4, 0.6]})
+    assert len(constraint.get_invalid(df)) == 0
+
+
+def test_linear_interpoint_constraint_get_invalid_violated():
+    """Test interpoint get_invalid returns full index when batch aggregate violates."""
+    constraint = ContinuousLinearConstraint(
+        parameters=["x"], operator="=", rhs=1.0, interpoint=True
+    )
+    df = pd.DataFrame({"x": [0.3, 0.6]})
+    assert list(constraint.get_invalid(df)) == list(df.index)
+
+
+def test_cardinality_constraint_get_invalid_raises():
+    """Test that ContinuousCardinalityConstraint.get_invalid raises NotImplementedError.
+
+    Row-level validation requires parameter bounds not stored on the constraint.
+    """
+    constraint = ContinuousCardinalityConstraint(
+        parameters=["x", "y", "z"], min_cardinality=1, max_cardinality=2
+    )
+    df = pd.DataFrame({"x": [0.0], "y": [1.0], "z": [0.0]})
+    with pytest.raises(NotImplementedError):
+        constraint.get_invalid(df)