MiniZinc Python - Pareto Front

pareto.py

# Main author: Jip J. Dekker <[email protected]>
# Copyright:   Jip J. Dekker, 2024
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

import asyncio
from enum import Enum
from typing import AsyncIterator, List, Tuple

from minizinc import Instance, Status, Result


class OptDirection(Enum):
    MAXIMIZE = 1
    MINIMIZE = 2

    def cmp_op(self) -> str:
        if self == OptDirection.MAXIMIZE:
            return ">"
        elif self == OptDirection.MINIMIZE:
            return "<"
        else:
            raise ValueError("Invalid optimization direction")

    def better(self, a, b) -> bool:
        if self == OptDirection.MAXIMIZE:
            return a > b
        elif self == OptDirection.MINIMIZE:
            return a < b
        else:
            raise ValueError("Invalid optimization direction")


async def pareto_solutions(
    inst: Instance, objectives: List[Tuple[str, OptDirection]], *args, **kwargs
) -> AsyncIterator[Result]:
    """
    Iterate soltutions of a multi-objective optimization problem where each solution is non-dominated by previous solutions.

    :param inst: The MiniZinc instance to solve
    :param objectives: A list of objectives given as tuples, where each tuple contains the identifier of an objective and its optimization direction
    :param args: Additional unnamed arguments to pass to the MiniZinc solve method
    :param kwargs: Additional named arguments to pass to the MiniZinc solve method
    """

    with inst.branch() as branch:
        result = await branch.solve_async(*args, **kwargs)
        while result.status.has_solution():
            yield result
            branch.add_string(
                "constraint "
                + "\\/".join(
                    [
                        f"({name} {o.cmp_op()} {result[name]})"
                        for (name, o) in objectives
                    ]
                )
                + ";"
            )
            result = await branch.solve_async(*args, **kwargs)


async def pareto_front(
    inst: Instance, objectives: List[Tuple[str, OptDirection]], *args, **kwargs
) -> List[Result]:
    """
    Find the pareto front of a multi-objective optimization problem.

    :param inst: The MiniZinc instance to solve
    :param objectives: A list of objectives given as tuples, where each tuple contains the identifier of an objective and its optimization direction
    :param args: Additional unnamed arguments to pass to the MiniZinc solve method
    :param kwargs: Additional named arguments to pass to the MiniZinc solve method
    """

    solns = []

    async for res in pareto_solutions(inst, objectives, *args, **kwargs):
        for i in range(len(solns)):
            if all(o.better(res[name], solns[i][name]) for (name, o) in objectives):
                solns[i] = res
                break
        else:
            solns.append(res)
    return solns


if __name__ == "__main__":
    from minizinc import Model, Solver

    model = Model()
    model.add_string(
        """
        include "all_different.mzn";

        var 1..4: x;
        var 1..4: y;
        var 1..4: z;

        constraint all_different([x, y, z]);
        solve satisfy;
        """
    )
    gecode = Solver.lookup("gecode")
    instance = Instance(gecode, model)

    objectives = [
        ("x", OptDirection.MAXIMIZE),
        ("y", OptDirection.MINIMIZE),
        ("z", OptDirection.MAXIMIZE),
    ]

    results = asyncio.run(pareto_front(instance, objectives))
    for res in results:
        print(res["x"], res["y"], res["z"])