XplorCplex

XplorCplex

XplorCplex(model: Model | None = None)

Bases: XplorModel[Model, Var, LinearExpr]

Xplor wrapper for the CPLEX solver using docplex.

This class provides a specialized wrapper for CPLEX, translating XplorModel's abstract operations into CPLEX-specific API calls via the docplex library for defining variables, constraints, optimizing, and extracting solutions.

Class Type Parameters:

Name	Bound or Constraints	Description	Default
ModelType	Model	The CPLEX model type.	required
ExpressionType	LinearExpr	Stores objective terms as CPLEX LinearExpr objects.	required

Attributes:

Name	Type	Description
model	Model	The instantiated CPLEX model object.

Initialize the XplorCplex model wrapper.

If no CPLEX model is provided, a new one is instantiated.

Parameters:

Name	Type	Description	Default
model	Model | None	An optional, pre-existing CPLEX model instance.	None

Source code in src/xplor/cplex/model.py

def __init__(self, model: Model | None = None) -> None:
    """Initialize the XplorCplex model wrapper.

    If no CPLEX model is provided, a new one is instantiated.

    Parameters
    ----------
    model : docplex.mp.model.Model | None, default None
        An optional, pre-existing CPLEX model instance.

    """
    model = Model() if model is None else model
    super().__init__(model=model)

Attributes

var cached property

var: _ProxyCplexVarExpr

The entry point for creating custom expression objects (VarExpr) that represent variables or columns used within a composite Polars expression chain.

This proxy acts similarly to polars.col(), allowing you to reference optimization variables (created via xmodel.add_vars()) or standard DataFrame columns in a solver-compatible expression.

The resulting expression object can be combined with standard Polars expressions to form constraints or objective function components.

Examples:

>>> xmodel = XplorCplex()
>>> df = df.with_columns(xmodel.add_vars("production"))
>>> df.select(total_cost = xmodel.var("production") * pl.col("cost"))

Functions

add_vars

add_vars(
    name: str,
    *,
    lb: float | str | Expr = 0.0,
    ub: float | str | Expr | None = None,
    obj: float | str | Expr | None = None,
    indices: Expr | list[str] | None = None,
    vtype: VariableType = "CONTINUOUS",
    priority: int | Expr = 0,
) -> Expr

Define and return a Var expression for optimization variables.

This method generates a Polars expression that, when consumed (e.g., via .with_columns()), creates optimization variables for every row and adds them to the underlying solver model.

Parameters:

Name	Type	Description	Default
name	str	The base name for the variables (e.g., "production" or "flow"). This name is used to retrieve variable values after optimization.	required
lb	float | str | Expr	Lower bound for created variables. Can be a scalar, a column name (str), or a Polars expression.	0.0
ub	float | str | Expr | None	Upper bound for created variables. If None, the solver default is used.	None
obj	float | str | Expr | None	Objective function coefficient for created variables. Can be a scalar, a column name, or a Polars expression.	None
indices	Expr | list[str] | None	Keys (column names) that uniquely identify each variable instance. Used to format the internal variable names (e.g., 'x[1,2]').	None
vtype	VariableType	The type of the variable (CONTINUOUS, INTEGER, or BINARY).	'CONTINUOUS'
priority	int | Expr	Multi-objective optimization priority. Higher priority numbers are optimized FIRST (priority 2 before priority 1 before priority 0). All objectives with the same priority are combined into a single weighted sum. Currently only supported by the Gurobi backend.	0

Returns:

Type	Description
Expr	A Polars expression (Var) that, when executed, adds variables to the model and returns them as an Object Series in the DataFrame.

Examples:

Assuming xmodel is an instance of a concrete class (XplorGurobi).

>>> # 1. Basic variable creation using columns for bounds:
>>> data = pl.DataFrame({"max_limit": [10.0, 5.0]})
>>> df = data.with_columns(
...     xmodel.add_vars("x", lb=0.0, ub=pl.col("max_limit"), obj=1.0)
... )
>>> # df["x"] now contains gurobipy.Var or mathopt.Variable objects.

>>> # s2. Creating integer variables indexed by two columns:
>>> data = pl.DataFrame({"time": [1, 1, 2, 2], "resource": ["A", "B", "A", "B"]})
>>> df = data.with_columns(
...     xmodel.add_vars(
...         "sched",
...         indices=["time", "resource"],
...         vtype=VarType.INTEGER,
...     )
... )
>>> # Variable names will look like 'sched[1,A]', 'sched[1,B]', etc.

Source code in src/xplor/model.py

def add_vars(
    self,
    name: str,
    *,
    lb: float | str | pl.Expr = 0.0,
    ub: float | str | pl.Expr | None = None,
    obj: float | str | pl.Expr | None = None,
    indices: pl.Expr | list[str] | None = None,
    vtype: VariableType = "CONTINUOUS",
    priority: int | pl.Expr = 0,
) -> pl.Expr:
    """Define and return a Var expression for optimization variables.

    This method generates a Polars expression that, when consumed (e.g., via
    `.with_columns()`), creates optimization variables for every row and adds
    them to the underlying solver model.

    Parameters
    ----------
    name : str
        The base name for the variables (e.g., "production" or "flow").
        This name is used to retrieve variable values after optimization.
    lb : float | str | pl.Expr, default 0.0
        Lower bound for created variables. Can be a scalar, a column name (str),
        or a Polars expression.
    ub : float | str | pl.Expr | None, default None
        Upper bound for created variables. If None, the solver default is used.
    obj: float | str | pl.Expr, default 0.0
        Objective function coefficient for created variables. Can be a scalar,
        a column name, or a Polars expression.
    indices: pl.Expr | None, default pl.row_index()
        Keys (column names) that uniquely identify each variable instance.
        Used to format the internal variable names (e.g., 'x[1,2]').
    vtype: VariableType
        The type of the variable (CONTINUOUS, INTEGER, or BINARY).
    priority: int | pl.Expr, default 0
        Multi-objective optimization priority. Higher priority numbers are optimized
        FIRST (priority 2 before priority 1 before priority 0). All objectives with
        the same priority are combined into a single weighted sum. Currently only
        supported by the Gurobi backend.

    Returns
    -------
    pl.Expr
        A Polars expression (`Var`) that, when executed, adds variables to the model
        and returns them as an `Object` Series in the DataFrame.

    Examples
    --------
    Assuming `xmodel` is an instance of a concrete class (`XplorGurobi`).

    >>> # 1. Basic variable creation using columns for bounds:
    >>> data = pl.DataFrame({"max_limit": [10.0, 5.0]})
    >>> df = data.with_columns(
    ...     xmodel.add_vars("x", lb=0.0, ub=pl.col("max_limit"), obj=1.0)
    ... )
    >>> # df["x"] now contains gurobipy.Var or mathopt.Variable objects.

    >>> # s2. Creating integer variables indexed by two columns:
    >>> data = pl.DataFrame({"time": [1, 1, 2, 2], "resource": ["A", "B", "A", "B"]})
    >>> df = data.with_columns(
    ...     xmodel.add_vars(
    ...         "sched",
    ...         indices=["time", "resource"],
    ...         vtype=VarType.INTEGER,
    ...     )
    ... )
    >>> # Variable names will look like 'sched[1,A]', 'sched[1,B]', etc.

    """
    indices = pl.concat_str(pl.row_index() if indices is None else indices, separator=",")
    return pl.map_batches(
        [
            parse_into_expr(lb).alias("lb"),
            parse_into_expr(ub).alias("ub"),
            parse_into_expr(obj).alias("obj"),
            parse_into_expr(priority).cast(pl.Int64).alias("priority"),
            pl.format(f"{name}[{{}}]", indices).alias("name"),
        ],
        lambda s: self._add_vars_wrapper(series_to_df(s), name=name, vtype=vtype),
        return_dtype=pl.Object,
    ).alias(name)

add_constrs

add_constrs(
    df: DataFrame,
    *constr_exprs: ConstrExpr,
    indices: Expr | list[str] | None = None,
    **named_constr_exprs: ConstrExpr,
) -> DataFrame

Define and return a Constr expression for model constraints.

This method accepts a symbolic relational expression (e.g., x <= 5) and generates a Polars expression that, when consumed (e.g., via .select()), adds the constraints to the underlying solver model.

The constraint is added row-wise if the input expression is a Series of expressions, or as a single constraint if the expression is aggregated (e.g., using .sum()).

Parameters:

Name	Type	Description	Default
df	DataFrame	The polars DataFrame used to create the constraints	required
constr_exprs	ConstrExpr	The constraints expression (e.g., a relational expression like xplor.var("x").sum() <= 10).	()
indices	Expr | list[str] | None	Keys (column names) that uniquely identify each constraint instance. Used to format the internal variable names (e.g., 'constr[1,2]').	None
named_constr_exprs	ConstrExpr	Other constraints expression	{}

Returns:

Type	Description
Expr	A Polars expression (Constr) that, when executed, adds constraints to the model and returns them as an Object Series in the DataFrame.
.. warning::	All constraints provided within a single call to add_constrs should have the same granularity (i.e., correspond to the same set of indices). If constraints with different granularities are provided, Polars' broadcasting mechanism might lead to constraints being added multiple times to the optimization model.

Examples:

Assuming df has been created and contains the variable Series df["x"].

>>> df.pipe(
...     xmodel.add_constrs,
        max_per_item = xplor.var("x") <= pl.col("capacity"),
        min_per_item = xplor.var("x") >= pl.col("min_threshold"),
        indices=["product"]
... )

Source code in src/xplor/model.py

def add_constrs(
    self,
    df: pl.DataFrame,
    *constr_exprs: ConstrExpr,
    indices: pl.Expr | list[str] | None = None,
    **named_constr_exprs: ConstrExpr,
) -> pl.DataFrame:
    r"""Define and return a Constr expression for model constraints.

    This method accepts a symbolic relational expression (e.g., `x <= 5`)
    and generates a Polars expression that, when consumed (e.g., via `.select()`),
    adds the constraints to the underlying solver model.

    The constraint is added row-wise if the input expression is a Series of
    expressions, or as a single constraint if the expression is aggregated
    (e.g., using `.sum()`).

    Parameters
    ----------
    df: pl.DataFrame
        The polars DataFrame used to create the constraints
    constr_exprs : ConstrExpr
        The constraints expression (e.g., a relational expression like
        `xplor.var("x").sum() <= 10`).
    indices: pl.Expr | None, default None
        Keys (column names) that uniquely identify each constraint instance.
        Used to format the internal variable names (e.g., 'constr[1,2]').
    named_constr_exprs : ConstrExpr
        Other constraints expression

    Returns
    -------
    pl.Expr
        A Polars expression (`Constr`) that, when executed, adds constraints
        to the model and returns them as an `Object` Series in the DataFrame.

    .. warning::
        All constraints provided within a single call to `add_constrs` should
        have the same granularity (i.e., correspond to the same set of indices).
        If constraints with different granularities are provided, Polars'
        broadcasting mechanism might lead to constraints being added multiple
        times to the optimization model.

    Examples
    --------
    Assuming `df` has been created and contains the variable Series `df["x"]`.

    >>> df.pipe(
    ...     xmodel.add_constrs,
            max_per_item = xplor.var("x") <= pl.col("capacity"),
            min_per_item = xplor.var("x") >= pl.col("min_threshold"),
            indices=["product"]
    ... )


    """
    if isinstance(indices, list):
        indices = pl.concat_str(indices, separator=",")

    for expr in constr_exprs:
        name = str(expr)
        assert name not in named_constr_exprs, f"Duplicated name for constraint {name}"
        named_constr_exprs[name] = expr

    # Process multi-output and single-output constraints separately
    exprs: list[pl.Expr] = []
    constrs_repr_d: dict[str, ExpressionRepr] = {}

    for name, expr in named_constr_exprs.items():
        if expr.meta.has_multiple_outputs():
            self._process_multi_output_constraint(df, expr, indices)
        else:
            constrs_repr_d[name], exprs = expr.parse(exprs)
    if constrs_repr_d:
        self._process_single_output_constraints(df, constrs_repr_d, exprs, indices)

    return df

minimize

minimize(
    df: DataFrame,
    /,
    priority: int = 0,
    **named_obj_exprs: VarExpr,
) -> DataFrame

Add minimization objectives to the model.

This method accepts named objective expressions and adds them to the model at the specified priority level. Multiple objectives at the same priority are combined into a single weighted sum.

Parameters:

Name	Type	Description	Default
df	DataFrame	The polars DataFrame used to evaluate the objective expressions.	required
priority	int	Multi-objective optimization priority. Higher priority numbers are optimized FIRST (priority 2 before priority 1 before priority 0). All objectives with the same priority are combined into a single weighted sum. Currently only supported by the Gurobi backend.	0
named_obj_exprs	Expr	Named objective expressions to minimize (e.g., sum_x = xplor.var("x").sum()).	{}

Returns:

Type	Description
DataFrame	The input DataFrame (unchanged), allowing for method chaining.

Examples:

Assuming df has been created and contains the variable Series df["x"].

>>> df.pipe(
...     xmodel.minimize,
...     total_cost = (xplor.var("x") * pl.col("cost")).sum(),
...     priority=1
... )

>>> # Multi-objective optimization
>>> df.pipe(
...     xmodel.minimize,
...     sum_x = xplor.var("x").sum(),
...     sum_y = xplor.var("y").sum(),
...     priority=2
... )

Source code in src/xplor/model.py

def minimize(
    self,
    df: pl.DataFrame,
    /,
    priority: int = 0,
    **named_obj_exprs: VarExpr,
) -> pl.DataFrame:
    """Add minimization objectives to the model.

    This method accepts named objective expressions and adds them to the model
    at the specified priority level. Multiple objectives at the same priority
    are combined into a single weighted sum.

    Parameters
    ----------
    df : pl.DataFrame
        The polars DataFrame used to evaluate the objective expressions.
    priority : int, default 0
        Multi-objective optimization priority. Higher priority numbers are optimized
        FIRST (priority 2 before priority 1 before priority 0). All objectives with
        the same priority are combined into a single weighted sum. Currently only
        supported by the Gurobi backend.
    named_obj_exprs : pl.Expr
        Named objective expressions to minimize (e.g., `sum_x = xplor.var("x").sum()`).

    Returns
    -------
    pl.DataFrame
        The input DataFrame (unchanged), allowing for method chaining.

    Examples
    --------
    Assuming `df` has been created and contains the variable Series `df["x"]`.

    >>> df.pipe(
    ...     xmodel.minimize,
    ...     total_cost = (xplor.var("x") * pl.col("cost")).sum(),
    ...     priority=1
    ... )

    >>> # Multi-objective optimization
    >>> df.pipe(
    ...     xmodel.minimize,
    ...     sum_x = xplor.var("x").sum(),
    ...     sum_y = xplor.var("y").sum(),
    ...     priority=2
    ... )

    """
    self._add_objectives(df, named_obj_exprs, priority, sense="minimize")
    return df

maximize

maximize(
    df: DataFrame,
    /,
    priority: int = 0,
    **named_obj_exprs: VarExpr,
) -> DataFrame

Add maximization objectives to the model.

This method accepts named objective expressions and adds them to the model at the specified priority level. Multiple objectives at the same priority are combined into a single weighted sum. Internally, maximization is converted to minimization by negating the objective coefficients.

Parameters:

Name	Type	Description	Default
df	DataFrame	The polars DataFrame used to evaluate the objective expressions.	required
priority	int	Multi-objective optimization priority. Higher priority numbers are optimized FIRST (priority 2 before priority 1 before priority 0). All objectives with the same priority are combined into a single weighted sum. Currently only supported by the Gurobi backend.	0
named_obj_exprs	Expr	Named objective expressions to maximize (e.g., sum_x = xplor.var("x").sum()).	{}

Returns:

Type	Description
DataFrame	The input DataFrame (unchanged), allowing for method chaining.

Examples:

Assuming df has been created and contains the variable Series df["x"].

>>> df.pipe(
...     xmodel.maximize,
...     total_revenue = (xplor.var("x") * pl.col("revenue")).sum(),
...     priority=1
... )

>>> # Multi-objective optimization
>>> df.pipe(
...     xmodel.maximize,
...     sum_x = xplor.var("x").sum(),
...     sum_y = xplor.var("y").sum(),
...     priority=2
... )

Source code in src/xplor/model.py

def maximize(
    self,
    df: pl.DataFrame,
    /,
    priority: int = 0,
    **named_obj_exprs: VarExpr,
) -> pl.DataFrame:
    """Add maximization objectives to the model.

    This method accepts named objective expressions and adds them to the model
    at the specified priority level. Multiple objectives at the same priority
    are combined into a single weighted sum. Internally, maximization is converted
    to minimization by negating the objective coefficients.

    Parameters
    ----------
    df : pl.DataFrame
        The polars DataFrame used to evaluate the objective expressions.
    priority : int, default 0
        Multi-objective optimization priority. Higher priority numbers are optimized
        FIRST (priority 2 before priority 1 before priority 0). All objectives with
        the same priority are combined into a single weighted sum. Currently only
        supported by the Gurobi backend.
    named_obj_exprs : pl.Expr
        Named objective expressions to maximize (e.g., `sum_x = xplor.var("x").sum()`).

    Returns
    -------
    pl.DataFrame
        The input DataFrame (unchanged), allowing for method chaining.

    Examples
    --------
    Assuming `df` has been created and contains the variable Series `df["x"]`.

    >>> df.pipe(
    ...     xmodel.maximize,
    ...     total_revenue = (xplor.var("x") * pl.col("revenue")).sum(),
    ...     priority=1
    ... )

    >>> # Multi-objective optimization
    >>> df.pipe(
    ...     xmodel.maximize,
    ...     sum_x = xplor.var("x").sum(),
    ...     sum_y = xplor.var("y").sum(),
    ...     priority=2
    ... )

    """
    self._add_objectives(df, named_obj_exprs, priority, sense="maximize")
    return df

optimize

optimize(**kwargs: Any) -> None

Solve the CPLEX model.

Before optimization, sets up multi-objective functions if multiple priority levels are defined. Higher priority values are optimized first.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Additional parameters passed to model.solve(). Common parameters include: - log_output: bool - Whether to display solver output - time_limit: float - Maximum solve time in seconds	{}

Source code in src/xplor/cplex/model.py

def optimize(self, **kwargs: Any) -> None:
    """Solve the CPLEX model.

    Before optimization, sets up multi-objective functions if multiple
    priority levels are defined. Higher priority values are optimized first.

    Parameters
    ----------
    **kwargs : Any
        Additional parameters passed to model.solve().
        Common parameters include:
        - log_output: bool - Whether to display solver output
        - time_limit: float - Maximum solve time in seconds

    """
    # Build multi-objective functions from accumulated terms
    if self._priority_obj_terms:
        # Sort priorities descending (highest user priority first)
        user_priorities = sorted(self._priority_obj_terms.keys(), reverse=True)

        if len(user_priorities) == 1:
            # Single objective
            priority = user_priorities[0]
            self.model.minimize(self._priority_obj_terms[priority])
        else:
            # Multi-objective optimization
            # CPLEX requires setting objectives with priorities
            for priority in user_priorities:
                expr = self._priority_obj_terms[priority]
                # Set as multi-objective with priority weight
                # Higher user priority gets higher CPLEX priority
                self.model.add_kpi(expr, publish_name=f"priority_{priority}")

            # For multi-objective in CPLEX, we minimize the primary (highest priority)
            # and the rest become constraints or are handled via blended objectives
            primary_priority = user_priorities[0]
            self.model.minimize(self._priority_obj_terms[primary_priority])

    # Solve the model
    self.solution = self.model.solve(**kwargs)

get_objective_value

get_objective_value() -> float

Return the objective value from the solved CPLEX model.

Returns:

Type	Description
float	The value of the objective function.

Raises:

Type	Description
ValueError	If the model has multiple objectives. Use get_multi_objective_values() instead.

Source code in src/xplor/cplex/model.py

def get_objective_value(self) -> float:
    """Return the objective value from the solved CPLEX model.

    Returns
    -------
    float
        The value of the objective function.

    Raises
    ------
    ValueError
        If the model has multiple objectives. Use get_multi_objective_values() instead.

    """
    if not self.solution:
        msg = "Model has not been optimized or no solution found."
        raise ValueError(msg)

    if len(self._priority_obj_terms) > 1:
        msg = (
            f"Model has {len(self._priority_obj_terms)} objectives. "
            "Use get_multi_objective_values() to retrieve all objective values."
        )
        raise ValueError(msg)

    return self.solution.get_objective_value()

get_multi_objective_values

get_multi_objective_values() -> dict[int, float]

Return all objective values from a multi-objective CPLEX model.

Returns a dictionary mapping user priority levels to their objective values.

Returns:

Type	Description
dict[int, float]	Dictionary mapping priority level to objective value. Keys are user priority levels (higher priority = higher number). Values are the objective values for each priority.

Examples:

>>> xmodel.optimize()
>>> obj_values = xmodel.get_multi_objective_values()
>>> print(obj_values)
{2: -150.0, 1: 50.0, 0: 10.0}  # priority -> objective value

Source code in src/xplor/cplex/model.py

def get_multi_objective_values(self) -> dict[int, float]:
    """Return all objective values from a multi-objective CPLEX model.

    Returns a dictionary mapping user priority levels to their objective values.

    Returns
    -------
    dict[int, float]
        Dictionary mapping priority level to objective value.
        Keys are user priority levels (higher priority = higher number).
        Values are the objective values for each priority.

    Examples
    --------
    >>> xmodel.optimize()
    >>> obj_values = xmodel.get_multi_objective_values()
    >>> print(obj_values)
    {2: -150.0, 1: 50.0, 0: 10.0}  # priority -> objective value

    """
    if not self.solution:
        msg = "Model has not been optimized or no solution found."
        raise ValueError(msg)

    if not self._priority_obj_terms:
        return {}

    # Evaluate each objective expression using the current solution
    result = {}
    for priority, expr in self._priority_obj_terms.items():
        # Get the value of the expression from the solution
        obj_value = expr.solution_value
        result[priority] = obj_value

    return result

read_values

read_values(name: Expr) -> Expr

Read the value of an optimization variable from the solution.

Parameters:

Name	Type	Description	Default
name	Expr	Expression to evaluate.	required

Returns:

Type	Description
Expr	Values of the variable expression.

Examples:

>>> xmodel: XplorModel
>>> df_with_solution = df.with_columns(xmodel.read_values(pl.selectors.object()))

Source code in src/xplor/cplex/model.py

def read_values(self, name: pl.Expr) -> pl.Expr:
    """Read the value of an optimization variable from the solution.

    Parameters
    ----------
    name : pl.Expr
        Expression to evaluate.

    Returns
    -------
    pl.Expr
        Values of the variable expression.

    Examples
    --------
    >>> xmodel: XplorModel
    >>> df_with_solution = df.with_columns(xmodel.read_values(pl.selectors.object()))

    """

    def _extract(v: Any) -> float | None:
        if v is None:
            return None
        if hasattr(v, "solution_value"):
            return v.solution_value
        return float(v)

    return name.map_batches(
        lambda d: cast_to_dtypes(
            pl.Series([_extract(v) for v in d]),
            self.var_types.get(d.name, "CONTINUOUS"),
        )
    )