XplorMathOpt

XplorMathOpt

XplorMathOpt(model: Model | None = None)

Bases: XplorModel[Model, Variable, LinearSum]

Xplor wrapper for the OR-Tools MathOpt solver.

This class extends XplorModel to provide an interface for building and solving optimization problems using OR-Tools MathOpt.

Class Type Parameters:

Name	Bound or Constraints	Description	Default
ModelType	Model	The MathOpt model type.	required
ExpressionType	LinearSum	Stores objective terms as MathOpt LinearSum expression objects.	required

Attributes:

Name	Type	Description
model	Model	The underlying OR-Tools MathOpt model instance.
result	SolveResult	The result object returned by MathOpt after optimization. It contains solution status, objective value, and variable values.

Initialize the XplorMathOpt model wrapper.

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

Parameters:

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

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

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

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

    Parameters
    ----------
    model : Model | None, default None
        An optional, pre-existing MathOpt model instance.

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

Attributes

var cached property

var: _ProxyVarExpr

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 = XplorMathOpt()
>>> 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(solver_type: SolverType | None = None) -> None

Solve the MathOpt model.

Uses mathopt.solve() to solve the model and stores the result internally.

Parameters:

Name	Type	Description	Default
solver_type	SolverType | None	The specific OR-Tools solver to use (e.g., GLOP, GSCIP). Defaults to MathOpt's native GLOP solver if none is provided.	SolverType.GLOP

Examples:

1. Using the default solver (GLOP):

   xmodel.optimize()

2. Specifying a different solver (requires setup/licensing for commercial solvers):

   from ortools.math_opt.python.parameters import SolverType xmodel.optimize(solver_type=SolverType.GUROBI)

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

def optimize(self, solver_type: SolverType | None = None) -> None:  # ty:ignore[invalid-method-override]
    """Solve the MathOpt model.

    Uses `mathopt.solve()` to solve the model and stores the result internally.

    Parameters
    ----------
    solver_type : SolverType | None, default SolverType.GLOP
        The specific OR-Tools solver to use (e.g., GLOP, GSCIP).
        Defaults to MathOpt's native GLOP solver if none is provided.

    Examples
    --------
    1. Using the default solver (GLOP):
       >>> xmodel.optimize()

    2. Specifying a different solver (requires setup/licensing for commercial solvers):
       >>> from ortools.math_opt.python.parameters import SolverType
       >>> xmodel.optimize(solver_type=SolverType.GUROBI)

    """
    solver_type = SolverType.GLOP if solver_type is None else solver_type

    # 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)
        max_user_priority = max(user_priorities)

        for user_priority in user_priorities:
            # Get the accumulated expression for this priority level
            obj_expr = self._priority_obj_terms[user_priority]

            # Invert priority: higher user priority → lower MathOpt priority
            mathopt_priority = max_user_priority - user_priority

            if mathopt_priority == 0:
                # Set as primary objective
                self.model.objective.set_to_linear_expression(obj_expr)
                self.model.objective.is_maximize = False
            else:
                # Add as auxiliary objective
                self.model.add_auxiliary_objective(
                    priority=mathopt_priority,
                    name=f"priority_{user_priority}",
                    expr=obj_expr,
                    is_maximize=False,
                )

    self.result = mathopt.solve(self.model, solver_type)

get_objective_value

get_objective_value() -> float

Return the objective value from the solved MathOpt model.

The value is read from the stored result object.

Returns:

Type	Description
float	The value of the objective function.

Raises:

Type	Description
Exception	If the model has not been optimized successfully (i.e., self.result is None).
ValueError	If the model has multiple objectives. Use get_multi_objective_values() instead.

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

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

    The value is read from the stored `result` object.

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

    Raises
    ------
    Exception
        If the model has not been optimized successfully (i.e., `self.result` is None).
    ValueError
        If the model has multiple objectives. Use get_multi_objective_values() instead.

    """
    if self.result is None:
        msg = "The model is not optimized."
        raise Exception(msg)

    # Check if model has multiple objectives
    num_aux_objs = len(list(self.model.auxiliary_objectives()))
    if num_aux_objs > 0:
        total_objs = 1 + num_aux_objs
        msg = (
            f"Model has {total_objs} objectives. "
            "Use get_multi_objective_values() to retrieve all objective values."
        )
        raise ValueError(msg)

    return self.result.objective_value()

get_multi_objective_values

get_multi_objective_values() -> dict[int, float]

Return all objective values from a multi-objective MathOpt 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(solver_type=mathopt.SolverType.GSCIP)
>>> 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/mathopt/model.py

def get_multi_objective_values(self) -> dict[int, float]:
    """Return all objective values from a multi-objective MathOpt 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(solver_type=mathopt.SolverType.GSCIP)
    >>> obj_values = xmodel.get_multi_objective_values()
    >>> print(obj_values)
    {2: -150.0, 1: 50.0, 0: 10.0}  # priority -> objective value

    """
    if self.result is None or not self._priority_obj_terms:
        return {}

    result_dict = {}
    user_priorities = sorted(self._priority_obj_terms.keys(), reverse=True)
    max_user_priority = max(user_priorities)

    # Primary objective (mathopt_priority=0) maps to highest user priority
    result_dict[max_user_priority] = self.result.objective_value()

    # Compute auxiliary objective values by evaluating expressions
    # MathOpt doesn't provide direct access to auxiliary objective values in results,
    # so we compute them from the stored expressions and variable values
    var_values = self.result.variable_values()

    for aux_obj in self.model.auxiliary_objectives():
        # Extract user priority from name "priority_{user_priority}"
        obj_name = aux_obj.name
        if obj_name.startswith("priority_"):
            user_priority = int(obj_name.split("_")[1])

            # Evaluate the auxiliary objective using variable values
            obj_value = aux_obj.offset
            for term in aux_obj.linear_terms():
                obj_value += term.coefficient * var_values.get(term.variable, 0.0)

            result_dict[user_priority] = obj_value

    return result_dict

read_values

read_values(name: Expr) -> Expr

Read the value of an optimization variable.

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/mathopt/model.py

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

    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()))

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