Implementing an OMMX Adapter

Implementing an OMMX Adapter#

As mentioned in Solve with multiple adapters and compare the results, OMMX Adapters have a common API. This common API is realized by inheriting the abstract base classes provided by the OMMX Python SDK. OMMX provides two abstract base classes depending on the type of adapter:

Solvers that produce multiple solutions can be automatically treated as solvers returning a single solution by selecting the best sample. Therefore, SamplerAdapter inherits SolverAdapter. If you are unsure which one to implement, consider the number of solutions: if the solver returns one solution, use SolverAdapter; if it returns multiple solutions, use SamplerAdapter. For example, exact solvers like PySCIPOpt should use SolverAdapter, while samplers like OpenJij should use SamplerAdapter.

In OMMX, a class inheriting ommx.adapter.SolverAdapter is called a Solver Adapter and one inheriting ommx.adapter.SamplerAdapter is called a Sampler Adapter. For clear explanation in this chapter, the software that the adapter wraps (such as PySCIPOpt or OpenJij) is referred to as “backend solver”.

Adapter Workflow#

The adapter process can be roughly divided into these 3 steps:

  1. Convert ommx.Instance into a format the backend solver can understand

  2. Run the backend solver to obtain a solution

  3. Convert the backend solver’s output into ommx.Solution or ommx.SampleSet

Because step 2 is nothing but the usage of the backend solver, we assume you are familiar with it. This tutorial explains steps 1 and 3.

Many backend solvers are designed to receive only the minimum necessary information to represent an optimization problem in a form suitable for their algorithms, whereas ommx.Instance contains more information, assuming optimization as part of data analysis. Therefore, step 1 involves discarding much of this information. Additionally, OMMX manages decision variables and constraints with IDs that are not necessarily sequential, while some backend solvers manage them by names or sequential numbers. This correspondence is needed in step 3, so the adapter must manage it.

Conversely, in step 3, ommx.Solution or ommx.SampleSet cannot be constructed solely from the backend solver’s output. Instead, the adapter will construct ommx.State or ommx.Samples from the backend solver’s output and the information from step 1, then convert it to ommx.Solution or ommx.SampleSet using ommx.Instance.

Implementing a Solver Adapter#

Here, we will implement a Solver Adapter using PySCIPOpt as an example. For a complete example, refer to ommx-pyscipopt-adapter.

For this tutorial, we will proceed in the following order to make it easier to execute step by step:

  • Implement functions to construct a PySCIPOpt model from ommx.Instance one by one.

  • Finally, combine these functions into the OMMXPySCIPOptAdapter class.

Custom Exception#

First, it is good to define custom exceptions. This makes it easier for users to understand which part is causing the problem when an exception occurs.

class OMMXPySCIPOptAdapterError(Exception):
    pass

OMMX can store a wide range of optimization problems, so the adapter declares the exact set it accepts with INPUT_CLASS, as shown later. The converter helpers below may still validate the representation they consume and raise while building solver input. Such converter or backend failures are not additional applicability conditions: adapter applicability is defined only by INPUT_CLASS membership.

Setting Decision Variables#

PySCIPOpt manages decision variables by name, so register the OMMX decision variable IDs as strings. This allows you to reconstruct ommx.State from PySCIPOpt decision variables in the decode_to_state function mentioned later. Note that the appropriate method depends on the backend solver’s implementation. The important thing is to retain the information needed to convert to ommx.State after obtaining the solution.

import pyscipopt
from ommx import (
    Constraint,
    DecisionVariable,
    Equality,
    Function,
    Instance,
    Kind,
    Sense,
    Solution,
    State,
)

def set_decision_variables(
    model: pyscipopt.Model,  # For tutorial purposes, we pass state as arguments, but managing with class is common
    instance: Instance
) -> dict[str, pyscipopt.Variable]:
    """
    Add decision variables to the model and create a mapping from variable names to variables
    """
    # Create PySCIPOpt variables from OMMX decision variable information
    for var in instance.used_decision_variables:
        if var.kind == Kind.Binary:
            model.addVar(name=str(var.id), vtype="B")
        elif var.kind == Kind.Integer:
            model.addVar(
                name=str(var.id), vtype="I", lb=var.bound.lower, ub=var.bound.upper
            )
        elif var.kind == Kind.Continuous:
            model.addVar(
                name=str(var.id), vtype="C", lb=var.bound.lower, ub=var.bound.upper
            )
        else:
            # Throw an error if an unsupported decision variable type is encountered
            raise OMMXPySCIPOptAdapterError(
                f"Unsupported decision variable kind: "
                f"id: {var.id}, kind: {var.kind}"
            )

    # If the objective is quadratic, add an auxiliary variable for linearization
    if instance.objective.degree() == 2:
        model.addVar(
            name="auxiliary_for_linearized_objective", vtype="C", lb=None, ub=None
        )

    # Create a dictionary to access the variables added to the model
    return {var.name: var for var in model.getVars()}

Converting ommx.Function to pyscipopt.Expr#

Implement a function to convert ommx.Function to pyscipopt.Expr. Since ommx.Function only has the OMMX decision variable IDs, you need to obtain the PySCIPOpt variables from the IDs using the variable name and variable mapping created in set_decision_variables.

def make_linear_expr(function: Function, varname_map: dict) -> pyscipopt.Expr:
    """Helper function to generate a linear expression"""
    return (
        pyscipopt.quicksum(
            coeff * varname_map[str(id)]
            for id, coeff in function.linear_terms.items()
        )
        + function.constant_term
    )


def make_quadratic_expr(function: Function, varname_map: dict) -> pyscipopt.Expr:
    """Helper function to generate a quadratic expression"""
    quad_terms = pyscipopt.quicksum(
        varname_map[str(row)] * varname_map[str(col)] * coeff
        for (row, col), coeff in function.quadratic_terms.items()
    )

    linear_terms = pyscipopt.quicksum(
        coeff * varname_map[str(var_id)]
        for var_id, coeff in function.linear_terms.items()
    )

    constant = function.constant_term

    return quad_terms + linear_terms + constant

Setting Objective Function and Constraints#

Add the objective function and constraints to the pyscipopt.Model. This part requires knowledge of what and how the backend solver supports. For example, in the following code, since PySCIPOpt cannot directly handle quadratic objective functions, an auxiliary variable is introduced according to the PySCIPOpt documentation.

import math

def set_objective(model: pyscipopt.Model, instance: Instance, varname_map: dict):
    """Set the objective function for the model"""
    objective = instance.objective

    if instance.sense == Sense.Maximize:
        sense = "maximize"
    elif instance.sense == Sense.Minimize:
        sense = "minimize"
    else:
        raise OMMXPySCIPOptAdapterError(
            f"Sense not supported: {instance.sense}"
        )

    degree = objective.degree()
    if degree == 0:
        model.setObjective(objective.constant_term, sense=sense)
    elif degree == 1:
        expr = make_linear_expr(objective, varname_map)
        model.setObjective(expr, sense=sense)
    elif degree == 2:
        # Since PySCIPOpt doesn't support quadratic objectives directly, linearize using an auxiliary variable
        auxilary_var = varname_map["auxiliary_for_linearized_objective"]

        # Set the auxiliary variable as the objective
        model.setObjective(auxilary_var, sense=sense)

        # Add a constraint for the auxiliary variable
        expr = make_quadratic_expr(objective, varname_map)
        if sense == "minimize":
            constr_expr = auxilary_var >= expr
        else:  # sense == "maximize"
            constr_expr = auxilary_var <= expr

        model.addCons(constr_expr, name="constraint_for_linearized_objective")
    else:
        raise OMMXPySCIPOptAdapterError(
            "The objective function must be `constant`, `linear`, or `quadratic`."
        )
        
def set_constraints(model: pyscipopt.Model, instance: Instance, varname_map: dict):
    """Set the constraints for the model"""
    # Process regular constraints. instance.constraints is a dict[int, Constraint]
    # keyed by constraint ID.
    for constraint_id, constraint in instance.constraints.items():
        # Generate an expression based on the type of constraint function
        f = constraint.function
        degree = f.degree()
        if degree == 0:
            # For constant constraints, check feasibility
            constant_value = f.constant_term
            if constraint.equality == Equality.EqualToZero and math.isclose(
                constant_value, 0, abs_tol=1e-6
            ):
                continue
            elif (
                constraint.equality == Equality.LessThanOrEqualToZero
                and constant_value <= 1e-6
            ):
                continue
            else:
                raise OMMXPySCIPOptAdapterError(
                    f"Infeasible constant constraint was found: id {constraint_id}"
                )
        elif degree == 1:
            expr = make_linear_expr(f, varname_map)
        elif degree == 2:
            expr = make_quadratic_expr(f, varname_map)
        else:
            raise OMMXPySCIPOptAdapterError(
                f"Constraints must be either `constant`, `linear` or `quadratic`. "
                f"id: {constraint_id}, "
                f"degree: {degree}"
            )

        # Add constraints based on the type (equality/inequality)
        if constraint.equality == Equality.EqualToZero:
            constr_expr = expr == 0
        elif constraint.equality == Equality.LessThanOrEqualToZero:
            constr_expr = expr <= 0
        else:
            raise OMMXPySCIPOptAdapterError(
                f"Not supported constraint equality: "
                f"id: {constraint_id}, equality: {constraint.equality}"
            )

        # Add the constraint to the model
        model.addCons(constr_expr, name=str(constraint_id))

Also, if the backend solver supports special constraints (e.g., SOS constraints), you need to add functions to handle them.

Now, we can construct a pycscipopt.Model from ommx.Instance.

Converting Obtained Solutions to ommx.State#

Next, implement a function to convert the solution obtained by solving the PySCIPOpt model to ommx.State. First, check if it is solved. SCIP has functions to guarantee optimality and detect unbounded solutions, so throw corresponding exceptions if detected. This also depends on the backend solver.

Warning

Note that ommx.adapter.InfeasibleDetected means that the optimization problem itself is infeasible, i.e., it is guaranteed to have no solutions. Do not use this when a heuristic solver fails to find any feasible solutions.

from ommx.adapter import InfeasibleDetected, UnboundedDetected

def decode_to_state(model: pyscipopt.Model, instance: Instance) -> State:
    """Create an ommx.State from an optimized PySCIPOpt Model"""
    if model.getStatus() == "unknown":
        raise OMMXPySCIPOptAdapterError(
            "The model may not be optimized. [status: unknown]"
        )

    if model.getStatus() == "infeasible":
        raise InfeasibleDetected("Model was infeasible")

    if model.getStatus() == "unbounded":
        raise UnboundedDetected("Model was unbounded")

    try:
        # Get the best solution
        sol = model.getBestSol()
        # Create a mapping from variable names to variables
        varname_map = {var.name: var for var in model.getVars()}
        # Create a State with a mapping from variable IDs to their values
        return State(
            entries={
                var.id: sol[varname_map[str(var.id)]]
                for var in instance.used_decision_variables
            }
        )
    except Exception:
        raise OMMXPySCIPOptAdapterError(
            f"There is no feasible solution. [status: {model.getStatus()}]"
        )

Creating a Class that Inherits ommx.adapter.SolverAdapter#

Finally, create a class that inherits ommx.adapter.SolverAdapter to standardize the API for each adapter. This is an abstract base class with @abstractmethod as follows:

from typing import ClassVar

class SolverAdapter(ABC):
    # Complete OMMX-defined condition for adapter applicability.
    INPUT_CLASS: ClassVar[InstanceClass]

    @classmethod
    def recommended_preparation_policy(cls) -> PreparationPolicy:
        return PreparationPolicy()

    @classmethod
    @abstractmethod
    def solve_without_preparation(
        cls,
        ommx_instance: Instance,
        *,
        diagnostics: DiagnosticsSink | None = None,
    ) -> Solution:
        pass

    @property
    @abstractmethod
    def solver_input(self) -> SolverInput:
        pass

    @abstractmethod
    def decode(self, data: SolverOutput) -> Solution:
        pass

This abstract base class assumes the following three use cases:

  • For the usual one-call workflow, use the inherited solve class method. It copies the input, applies recommended_preparation_policy() for INPUT_CLASS, and calls solve_without_preparation on that temporary prepared copy.

  • If the application prepares an exact Adapter input itself, call solve_without_preparation. This method performs no Preparation.

  • If you adjust the backend solver’s parameters, use solver_input to get the data structure for the backend solver (in this case, pyscipopt.Model), adjust it, then input it to the backend solver, and finally convert the backend solver’s output using decode.

The inherited solve method covers Adapters with no additional options. A concrete Adapter declares each Adapter-specific option with an explicit typed signature on the methods that support it. If an option has the same meaning before and after Preparation, the Adapter may declare it on both methods and forward it explicitly. If an option depends on the exact prepared input and the Adapter defines no way to transport it through Preparation, the Adapter may instead expose it only on solve_without_preparation. Do not use a catch-all **kwargs: it prevents type checkers from rejecting unknown options. The reserved diagnostics keyword is owned by Run.log_solve. When Run.log_solve(..., store_diagnostics=True) is used, adapters may record adapter-defined diagnostic reports into that sink; None means diagnostics are disabled.

Solving a Knapsack Problem Using the Solver Adapter#

For verification, let’s solve a knapsack problem using this.

v = [10, 13, 18, 31, 7, 15]
w = [11, 25, 20, 35, 10, 33]
W = 47
N = len(v)

x = [
    DecisionVariable.binary(
        id=i,
        name="x",
        subscripts=[i],
    )
    for i in range(N)
]
instance = Instance.from_components(
    decision_variables=x,
    objective=sum(v[i] * x[i] for i in range(N)),
    constraints={0: sum(w[i] * x[i] for i in range(N)) - W <= 0},
    sense=Sense.Maximize,
)

solution = OMMXPySCIPOptAdapter.solve(instance)

Implementing a Sampler Adapter#

Next, let’s create a Sampler Adapter using OpenJij. OpenJij includes openjij.SASampler for Simulated Annealing (SA) and openjij.SQASampler for Simulated Quantum Annealing (SQA). In this tutorial, we will use SASampler as an example.

For simplicity, this tutorial omits the parameters passed to OpenJij. For more details, refer to the implementation of ommx-openjij-adapter. For how to use the OpenJij Adapter, refer to Sampling from QUBO with OMMX Adapter.

Converting openjij.Response to ommx.Samples#

OpenJij manages decision variables with IDs that are not necessarily sequential, similar to OMMX, so there is no need to create an ID correspondence table as in the case of PySCIPOpt.

The sample results from OpenJij are obtained as openjij.Response, so implement a function to convert this to ommx.Samples. OpenJij returns the number of occurrences of the same sample as num_occurrence. On the other hand, ommx.Samples has unique sample IDs for each sample, and the same value samples are compressed as SamplesEntry. Note that a conversion is needed to bridge this difference.

import openjij as oj
from ommx import Instance, SampleSet, Samples, Sense, Solution, State

def decode_to_samples(response: oj.Response) -> Samples:
    # Generate sample IDs
    samples = Samples({})  # Create empty samples
    sample_id = 0

    num_reads = len(response.record.num_occurrences)
    for i in range(num_reads):
        sample = response.record.sample[i]
        state = State(entries=zip(response.variables, sample))
        # `num_occurrences` is encoded into sample ID list.
        # For example, if `num_occurrences` is 2, there are two samples with the same state, thus two sample IDs are generated.
        ids = []
        for _ in range(response.record.num_occurrences[i]):
            ids.append(sample_id)
            sample_id += 1
        samples.append(ids, state)

    return samples

Note that at this stage, ommx.Instance or its extracted correspondence table is not needed because there is no need to consider ID correspondence.

Implementing a Class that Inherits ommx.adapter.SamplerAdapter#

In the case of PySCIPOpt, we inherited SolverAdapter, but this time we will inherit SamplerAdapter. This has three @abstractmethod as follows:

class SamplerAdapter(SolverAdapter):
    @classmethod
    @abstractmethod
    def sample_without_preparation(
        cls,
        ommx_instance: Instance,
        *,
        diagnostics: DiagnosticsSink | None = None,
    ) -> SampleSet:
        pass

    @property
    @abstractmethod
    def sampler_input(self) -> SamplerInput:
        pass

    @abstractmethod
    def decode_to_sampleset(self, data: SamplerOutput) -> SampleSet:
        pass

SamplerAdapter supplies the same two-level API. The inherited sample copies the caller’s Instance and prepares that copy, while sample_without_preparation receives the exact Adapter input. SamplerAdapter.solve_without_preparation selects best_feasible from sample_without_preparation; solver_input delegates to sampler_input, and decode returns decode_to_sampleset(...).best_feasible. A Sampler implementation therefore defines only its sampler-owned sample_without_preparation, sampler_input, and decode_to_sampleset operations when it has no additional options. A Sampler with Adapter-specific options declares each option with explicit typed signatures on the methods that support it. Options whose meaning survives Preparation may use matching signatures on sample and sample_without_preparation. When an option depends on the exact prepared sampler input and no transport through Preparation is defined, a concrete Sampler may expose it only on sample_without_preparation. The same choice is available when exposing options through the Solver API.

As with solve, the reserved diagnostics keyword is owned by Run.log_sample. A sampler may record adapter-defined reports into the sink when it is not None.

from ommx.adapter import DiagnosticsSink, SamplerAdapter

class OMMXOpenJijSAAdapter(SamplerAdapter):
    """
    Sampling QUBO with Simulated Annealing (SA) by `openjij.SASampler`
    """

    INPUT_CLASS = InstanceClass(
        [
            InstanceClassClause(
                label="tutorial-binary-qubo",
                allowed_variable_kinds={Kind.Binary},
                objective_polynomial_requirement=PolynomialRequirement.at_most(2),
                allowed_senses={Sense.Minimize},
            )
        ]
    )

    # Retain the Instance because it is required to convert to SampleSet
    ommx_instance: Instance
    
    def __init__(self, ommx_instance: Instance):
        self.require_applicable(ommx_instance)
        self.ommx_instance = ommx_instance

    # Perform sampling
    def _sample(self) -> oj.Response:
        sampler = oj.SASampler()
        # Convert to QUBO dictionary format
        # QUBO conversion can fail here even after applicability was established.
        # This is a converter error, not an applicability result.
        qubo, _offset = self.ommx_instance.as_qubo_format()
        return sampler.sample_qubo(qubo)

    # Common method for performing sampling
    @classmethod
    def sample_without_preparation(
        cls,
        ommx_instance: Instance,
        *,
        diagnostics: DiagnosticsSink | None = None,
    ) -> SampleSet:
        _ = diagnostics
        adapter = cls(ommx_instance)
        response = adapter._sample()
        return adapter.decode_to_sampleset(response)
    
    # In this adapter, `SamplerInput` uses a QUBO dictionary
    @property
    def sampler_input(self) -> dict[tuple[int, int], float]:
        qubo, _offset = self.ommx_instance.as_qubo_format()
        return qubo
   
    # Convert OpenJij Response to a SampleSet
    def decode_to_sampleset(self, data: oj.Response) -> SampleSet:
        samples = decode_to_samples(data)
        # The information stored in `ommx.Instance` is required here
        return self.ommx_instance.evaluate_samples(samples)

Sampling using our Adapter#

Let’s sample from the following QUBO using our Adapter:

\[ \begin{aligned} \min & \quad -x_0 - x_1 + 2 x_0 x_1 \\ & \quad x_0, x_1 \in \{0, 1\} \end{aligned} \]
x = [DecisionVariable.binary(id, name="x", subscripts=[id]) for id in range(2)]
instance = Instance.from_components(
    decision_variables=x,
    objective=-x[0] - x[1] + 2 * x[0] * x[1],
    constraints={},
    sense=Sense.Minimize,
)

sample_set = OMMXOpenJijSAAdapter.sample(instance)
sample_set.summary

Summary#

In this tutorial, we learned how to implement an OMMX Adapter by connecting to PySCIPOpt as a Solver Adapter and OpenJij as a Sampler Adapter. Here are the key points when implementing an OMMX Adapter:

  1. Implement an OMMX Adapter by inheriting the abstract base class SolverAdapter or SamplerAdapter.

  2. Define applicability with INPUT_CLASS and implement the preparation-free solve_without_preparation() or sample_without_preparation() method. The usual solve() and sample() APIs copy the caller’s Instance and apply the fresh policy from recommended_preparation_policy() automatically. Applications customize Preparation by preparing their Instance and invoking the preparation-free API.

  3. The main steps of the implementation are as follows:

    • Convert ommx.Instance into a format that the backend solver can understand.

    • Run the backend solver to obtain a solution.

    • Convert the backend solver’s output into ommx.Solution or ommx.SampleSet.

  4. Keep converter-local and backend validation in the solver-input construction path, and report its failures as conversion or backend errors rather than applicability failures.

  5. Pay attention to managing IDs and mapping variables to bridge the backend solver and OMMX.

If you want to connect your own backend solver to OMMX, refer to this tutorial for implementation. By implementing an OMMX Adapter following this tutorial, you can use optimization with various backend solvers through a common API.

For more detailed implementation examples, refer to the repositories such as ommx-pyscipopt-adapter and ommx-openjij-adapter.

Contents