From ce504a7691ae51016078257555bf7b48a854b0e2 Mon Sep 17 00:00:00 2001 From: OrestisLomis Date: Wed, 6 May 2026 11:59:45 +0200 Subject: [PATCH 1/8] add optalcp solver with some workarounds for masked values --- cpmpy/exceptions.py | 7 +- cpmpy/expressions/variables.py | 10 + cpmpy/model.py | 8 + cpmpy/solvers/__init__.py | 3 + cpmpy/solvers/optalcp.py | 436 +++++++++++++++++++++++++++++++++ cpmpy/solvers/utils.py | 2 + tests/test_solverinterface.py | 21 +- 7 files changed, 484 insertions(+), 3 deletions(-) create mode 100644 cpmpy/solvers/optalcp.py diff --git a/cpmpy/exceptions.py b/cpmpy/exceptions.py index 07e9c7b8b..0cf9f4732 100644 --- a/cpmpy/exceptions.py +++ b/cpmpy/exceptions.py @@ -42,4 +42,9 @@ class GCSVerificationException(CPMpyException): class TransformationNotImplementedError(CPMpyException): '''Raised when a transformation is not implemented for a certain expression''' - pass \ No newline at end of file + pass + + +class MaskedSolverValueError(CPMpyException): + '''Raised when a solver found a solution but intentionally masked decision-variable values''' + pass diff --git a/cpmpy/expressions/variables.py b/cpmpy/expressions/variables.py index 80c93b36e..9f31862b0 100644 --- a/cpmpy/expressions/variables.py +++ b/cpmpy/expressions/variables.py @@ -65,6 +65,7 @@ import numpy as np import cpmpy as cp # to avoid circular import +from ..exceptions import MaskedSolverValueError from .core import Expression, ExprLike, BoolExprLike, ListLike, BoolVal from .utils import is_num, is_int, is_boolexpr, get_bounds @@ -327,6 +328,7 @@ def __init__(self, lb: int, ub: int, name: str): self.ub = ub self.name = name self._value: Optional[int] = None + self._value_error: Optional[str] = None def has_subexpr(self) -> bool: """Does it contains nested Expressions? @@ -343,6 +345,8 @@ def value(self) -> Optional[int]: """ the value obtained in the last solve call (or 'None') """ + if self._value_error is not None: + raise MaskedSolverValueError(self._value_error) return self._value def get_bounds(self) -> tuple[int, int]: @@ -353,6 +357,12 @@ def clear(self) -> None: """ clear the value obtained from the last solve call """ self._value = None + self._value_error = None + + def _mask_value(self, message: str) -> None: + """Mark the last solver value as intentionally unavailable.""" + self._value = None + self._value_error = message def __repr__(self) -> str: return self.name diff --git a/cpmpy/model.py b/cpmpy/model.py index e07b7d371..127fe1c87 100644 --- a/cpmpy/model.py +++ b/cpmpy/model.py @@ -57,6 +57,7 @@ def __init__(self, *args, minimize=None, maximize=None): """ assert ((minimize is None) or (maximize is None)), "can not set both minimize and maximize" self.cpm_status = SolverStatus("Model") # status of solving this model, will be replaced + self.objective_value_ = None # init list of constraints and objective self.constraints = [] @@ -147,6 +148,7 @@ def objective(self, expr, minimize): """ self.objective_ = expr self.objective_is_min = minimize + self.objective_value_ = None def has_objective(self): """ @@ -164,6 +166,10 @@ def objective_value(self): Returns: int, optional: The objective value as an integer or ``None`` if it is not run or is a satisfaction problem """ + if self.objective_ is None: + return None + if self.objective_value_ is not None: + return self.objective_value_ return self.objective_.value() def solve(self, solver:Optional[str]=None, time_limit:Optional[int|float]=None, **kwargs): @@ -197,6 +203,7 @@ def solve(self, solver:Optional[str]=None, time_limit:Optional[int|float]=None, ret = s.solve(time_limit=time_limit, **kwargs) # store CPMpy status (s object has no further use) self.cpm_status = s.status() + self.objective_value_ = s.objective_value() return ret def solveAll(self, solver:Optional[str]=None, display:Optional[Callback]=None, time_limit:Optional[int|float]=None, solution_limit:Optional[int]=None, **kwargs): @@ -227,6 +234,7 @@ def solveAll(self, solver:Optional[str]=None, display:Optional[Callback]=None, t ret = s.solveAll(display=display,time_limit=time_limit,solution_limit=solution_limit, call_from_model=True, **kwargs) # store CPMpy status (s object has no further use) self.cpm_status = s.status() + self.objective_value_ = s.objective_value() return ret def status(self): diff --git a/cpmpy/solvers/__init__.py b/cpmpy/solvers/__init__.py index a888927bb..b82079832 100644 --- a/cpmpy/solvers/__init__.py +++ b/cpmpy/solvers/__init__.py @@ -42,6 +42,7 @@ pumpkin cplex hexaly + optalcp rc2 scip @@ -78,6 +79,7 @@ from .pumpkin import CPM_pumpkin from .cplex import CPM_cplex from .hexaly import CPM_hexaly +from .optalcp import CPM_optalcp from .rc2 import CPM_rc2 from .scip import CPM_scip @@ -90,6 +92,7 @@ "CPM_gurobi", "CPM_hexaly", "CPM_minizinc", + "CPM_optalcp", "CPM_ortools", "CPM_pindakaas", "CPM_pumpkin", diff --git a/cpmpy/solvers/optalcp.py b/cpmpy/solvers/optalcp.py new file mode 100644 index 000000000..18585d195 --- /dev/null +++ b/cpmpy/solvers/optalcp.py @@ -0,0 +1,436 @@ +""" + Interface to OptalCP's Python API. + + OptalCP is a scheduling-oriented constraint programming solver with a native Python API. + + Always use :func:`cp.SolverLookup.get("optalcp") ` to instantiate the solver object. + + ============ + Installation + ============ + + Requires the `optalcp` Python package and an OptalCP solver binary. + + Public preview installation currently works via: + + .. code-block:: console + + $ pip install git+https://github.com/ScheduleOpt/optalcp-py-bin-preview@latest + + The preview edition solves models and reports objective values, but masks decision-variable assignments. + Academic and full editions expose variable values as well. + + Documentation: + https://optalcp.com/docs/ + + =============== + List of classes + =============== + + .. autosummary:: + :nosignatures: + + CPM_optalcp +""" + +from typing import Optional + +from .solver_interface import SolverInterface, SolverStatus, ExitStatus, Callback +from .. import DirectConstraint +from ..exceptions import MaskedSolverValueError, NotSupportedError +from ..expressions.core import Comparison, Operator, BoolVal +from ..expressions.globalconstraints import GlobalConstraint +from ..expressions.globalfunctions import GlobalFunction +from ..expressions.variables import _BoolVarImpl, NegBoolView, _IntVarImpl, _NumVarImpl, intvar +from ..expressions.utils import is_num, is_any_list, eval_comparison, get_bounds, get_nonneg_args, implies +from ..transformations.get_variables import get_variables +from ..transformations.normalize import toplevel_list +from ..transformations.decompose_global import decompose_in_tree, decompose_objective +from ..transformations.safening import no_partial_functions + + +class CPM_optalcp(SolverInterface): + """ + Interface to OptalCP's Python API. + + Creates the following attributes (see parent constructor for more): + + - ``opt_model``: object, OptalCP model object + """ + + supported_global_constraints = frozenset({ + "cumulative", "cumulative_optional", "no_overlap", "no_overlap_optional", + "min", "max", "abs", "mul", "div", + }) + supported_reified_global_constraints = frozenset() + + _masked_value_message = ( + "OptalCP Preview solved the model but masks decision-variable values. " + "Install an Academic or Full OptalCP edition to retrieve variable assignments." + ) + + @staticmethod + def installed(): + try: + import optalcp + return True + except ModuleNotFoundError: + return False + + @staticmethod + def supported(): + if not CPM_optalcp.installed(): + return False + try: + import optalcp + optalcp.Solver.find_solver({}) + return True + except Exception: + return False + + @staticmethod + def version() -> Optional[str]: + try: + import optalcp + return optalcp.__version__ + except ModuleNotFoundError: + return None + + def __init__(self, cpm_model=None, subsolver=None): + if not self.installed(): + raise ModuleNotFoundError( + "CPM_optalcp: Install the 'optalcp' Python package and an OptalCP binary to use this solver interface." + ) + + import optalcp + + assert subsolver is None + self.opt_model = optalcp.Model() + self._objective_is_min = None + self._values_masked = False + super().__init__(name="optalcp", cpm_model=cpm_model) + + @property + def native_model(self): + return self.opt_model + + def solve(self, time_limit: Optional[float] = None, **kwargs): + self.solver_vars(list(self.user_vars)) + + if not len(self.user_vars): + self.add(intvar(1, 1) == 1) + + if "printLog" not in kwargs: + kwargs["printLog"] = False + + if time_limit is not None: + if time_limit <= 0: + raise ValueError("Time limit must be positive") + kwargs["timeLimit"] = time_limit + + self.opt_result = self.opt_model.solve(kwargs or None) + + self.cpm_status = SolverStatus(self.name) + self.cpm_status.runtime = self.opt_result.duration + + if self.opt_result.nb_solutions == 0: + self.cpm_status.exitstatus = ExitStatus.UNSATISFIABLE if self.opt_result.proof else ExitStatus.UNKNOWN + elif self.has_objective(): + self.cpm_status.exitstatus = ExitStatus.OPTIMAL if self.opt_result.proof else ExitStatus.FEASIBLE + else: + self.cpm_status.exitstatus = ExitStatus.FEASIBLE + + has_sol = self._solve_return(self.cpm_status) + self.objective_value_ = None + self._values_masked = False + + if has_sol: + self._populate_values(self.opt_result.solution) + if self.has_objective(): + self.objective_value_ = self.opt_result.objective + else: + for cpm_var in self.user_vars: + cpm_var.clear() + + return has_sol + + def solveAll(self, display: Optional[Callback] = None, time_limit: Optional[float] = None, + solution_limit: Optional[int] = None, call_from_model=False, **kwargs): + try: + return super().solveAll(display=display, time_limit=time_limit, + solution_limit=solution_limit, call_from_model=call_from_model, **kwargs) + except MaskedSolverValueError as exc: + raise NotSupportedError( + "OptalCP solveAll() requires visible decision-variable values. " + "The Preview edition masks them." + ) from exc + + def _populate_values(self, solution): + if solution is None: + for cpm_var in self.user_vars: + cpm_var.clear() + return + + masked = False + values = {} + for cpm_var in self.user_vars: + opt_var = self.solver_var(cpm_var) + value = solution.get_value(opt_var) + values[cpm_var] = value + if value is None: + masked = True + + if masked: + self._values_masked = True + for cpm_var in self.user_vars: + cpm_var._mask_value(self._masked_value_message) + return + + for cpm_var, value in values.items(): + cpm_var.clear() + if isinstance(cpm_var, _BoolVarImpl): + cpm_var._value = bool(value) + else: + cpm_var._value = value + + def solver_var(self, cpm_var): + if is_num(cpm_var): + return cpm_var + + if isinstance(cpm_var, NegBoolView): + return ~self.solver_var(cpm_var._bv) + + if cpm_var not in self._varmap: + if isinstance(cpm_var, _BoolVarImpl): + revar = self.opt_model.bool_var(name=str(cpm_var)) + elif isinstance(cpm_var, _IntVarImpl): + revar = self.opt_model.int_var(min=cpm_var.lb, max=cpm_var.ub, name=str(cpm_var)) + else: + raise NotImplementedError(f"Not a known var {cpm_var}") + self._varmap[cpm_var] = revar + + return self._varmap[cpm_var] + + def objective(self, expr, minimize=True): + get_variables(expr, self.user_vars) + + obj, decomp_cons = decompose_objective( + expr, + supported=self.supported_global_constraints, + supported_reified=self.supported_reified_global_constraints, + csemap=self._csemap + ) + self.add(decomp_cons) + + opt_obj = self._opt_expr(obj) + self._objective_is_min = minimize + if minimize: + self.opt_model.minimize(opt_obj) + else: + self.opt_model.maximize(opt_obj) + + def has_objective(self): + return self._objective_is_min is not None + + def transform(self, cpm_expr): + cpm_cons = toplevel_list(cpm_expr) + cpm_cons = no_partial_functions(cpm_cons, safen_toplevel=frozenset({})) + cpm_cons = decompose_in_tree( + cpm_cons, + supported=self.supported_global_constraints, + supported_reified=self.supported_reified_global_constraints, + csemap=self._csemap + ) + return cpm_cons + + def add(self, cpm_expr): + get_variables(cpm_expr, collect=self.user_vars) + + for cpm_con in self.transform(cpm_expr): + opt_con = self._opt_expr(cpm_con) + if is_any_list(opt_con): + self.opt_model.enforce(opt_con) + else: + self.opt_model.enforce(opt_con) + + return self + + __add__ = add + + def _opt_expr(self, cpm_con): + if is_any_list(cpm_con): + return [self._opt_expr(con) for con in cpm_con] + + if isinstance(cpm_con, BoolVal): + return cpm_con.args[0] + + if is_num(cpm_con): + return cpm_con + + if isinstance(cpm_con, _NumVarImpl): + return self.solver_var(cpm_con) + + if isinstance(cpm_con, Operator): + if cpm_con.name == "and": + args = self._opt_expr(cpm_con.args) + if not args: + return True + cur = args[0] + for arg in args[1:]: + cur = self.opt_model.and_(cur, arg) + return cur + if cpm_con.name == "or": + args = self._opt_expr(cpm_con.args) + if not args: + return False + cur = args[0] + for arg in args[1:]: + cur = self.opt_model.or_(cur, arg) + return cur + if cpm_con.name == "->": + lhs, rhs = self._opt_expr(cpm_con.args) + return self.opt_model.implies(lhs, rhs) + if cpm_con.name == "not": + return self.opt_model.not_(self._opt_expr(cpm_con.args[0])) + if cpm_con.name == "sum": + return self.opt_model.sum(self._opt_expr(cpm_con.args)) + if cpm_con.name == "wsum": + weights = cpm_con.args[0] + exprs = self._opt_expr(cpm_con.args[1]) + return self.opt_model.sum([w * x for w, x in zip(weights, exprs)]) + if cpm_con.name == "sub": + lhs, rhs = self._opt_expr(cpm_con.args) + return lhs - rhs + if cpm_con.name == "-": + return -self._opt_expr(cpm_con.args[0]) + raise NotImplementedError(f"Operator {cpm_con} not implemented for OptalCP") + + if isinstance(cpm_con, Comparison): + lhs, rhs = self._opt_expr(cpm_con.args) + return eval_comparison(cpm_con.name, lhs, rhs) + + if isinstance(cpm_con, GlobalConstraint): + if cpm_con.name in ("cumulative", "cumulative_optional"): + return self._opt_cumulative(cpm_con) + if cpm_con.name == "no_overlap": + if len(cpm_con.args) == 2: + start, dur = cpm_con.args + end = None + else: + start, dur, end = cpm_con.args + tasks, cons = self._make_tasks(start, dur, end, None) + task_list = [task for task in tasks if task is not None] + if len(task_list) > 1: + cons.append(self.opt_model.no_overlap(task_list)) + return cons + if cpm_con.name == "no_overlap_optional": + if len(cpm_con.args) == 3: + start, dur, is_present = cpm_con.args + end = None + else: + start, dur, end, is_present = cpm_con.args + tasks, cons = self._make_tasks(start, dur, end, is_present) + task_list = [task for task in tasks if task is not None] + if len(task_list) > 1: + cons.append(self.opt_model.no_overlap(task_list)) + return cons + if isinstance(cpm_con, DirectConstraint): + return cpm_con.callSolver(self, self.opt_model) + raise NotImplementedError(f"Global constraint {cpm_con} not supported by OptalCP backend") + + if isinstance(cpm_con, GlobalFunction): + if cpm_con.name == "min": + return self.opt_model.min(self._opt_expr(cpm_con.args)) + if cpm_con.name == "max": + return self.opt_model.max(self._opt_expr(cpm_con.args)) + if cpm_con.name == "abs": + return self.opt_model.abs(self._opt_expr(cpm_con.args)[0]) + if cpm_con.name == "mul": + lhs, rhs = self._opt_expr(cpm_con.args) + return lhs * rhs + if cpm_con.name == "div": + lhs, rhs = self._opt_expr(cpm_con.args) + return lhs // rhs + raise NotImplementedError(f"Global function {cpm_con} not supported by OptalCP backend") + + raise NotImplementedError("OptalCP: constraint not (yet) supported", cpm_con) + + def _opt_cumulative(self, cpm_con): + if cpm_con.name == "cumulative": + is_present = None + if len(cpm_con.args) == 4: + start, dur, demand, capacity = cpm_con.args + end = None + else: + start, dur, end, demand, capacity = cpm_con.args + else: + if len(cpm_con.args) == 5: + start, dur, demand, capacity, is_present = cpm_con.args + end = None + else: + start, dur, end, demand, capacity, is_present = cpm_con.args + + tasks, cons = self._make_tasks(start, dur, end, is_present) + demand_lst, demand_cons = get_nonneg_args(demand, is_present) + cons += self._opt_expr(demand_cons) + + pulses = [] + for task, h in zip(tasks, demand_lst): + if task is not None: + pulses.append(task.pulse(self._opt_expr(h))) + + if pulses: + cons.append(self.opt_model.sum(pulses) <= self._opt_expr(capacity)) + return cons + + def _make_tasks(self, start, dur, end, is_present): + if end is None: + end = [None for _ in range(len(start))] + + dur, dur_cons = get_nonneg_args(dur, is_present) + extra_cons = self._opt_expr(dur_cons) + + if is_present is None: + is_present = [None] * len(start) + + tasks = [] + for s, d, e, p in zip(start, dur, end, is_present): + task, task_cons = self._make_task(s, d, e, p) + tasks.append(task) + extra_cons += task_cons + return tasks, extra_cons + + def _make_task(self, start, dur, end, is_present): + assert get_bounds(dur)[0] >= 0, "optalcp does not support intervals with negative duration, use `utils.get_nonneg_args` first" + + is_optional = is_present is not None + if not is_optional: + is_present = BoolVal(True) + + lb, ub = get_bounds(dur) + extra_cons = [] + if lb == 0 == ub: + if end is None: + return None, [] + return None, self._opt_expr([implies(is_present, start == end)]) + + task = self.opt_model.interval_var( + start=get_bounds(start), + end=get_bounds(end) if end is not None else None, + length=get_bounds(dur), + optional=is_optional + ) + + if is_optional: + extra_cons.append(task.presence() == self.solver_var(is_present)) + extra_cons.append(self.opt_model.implies(task.presence(), task.start() == self._opt_expr(start))) + extra_cons.append(self.opt_model.implies(task.presence(), task.length() == self._opt_expr(dur))) + if end is not None: + extra_cons.append(self.opt_model.implies(task.presence(), task.end() == self._opt_expr(end))) + else: + extra_cons.append(task.start() == self._opt_expr(start)) + extra_cons.append(task.length() == self._opt_expr(dur)) + if end is not None: + extra_cons.append(task.end() == self._opt_expr(end)) + + return task, extra_cons diff --git a/cpmpy/solvers/utils.py b/cpmpy/solvers/utils.py index a527ac8ca..e0fed03f3 100644 --- a/cpmpy/solvers/utils.py +++ b/cpmpy/solvers/utils.py @@ -33,6 +33,7 @@ from .cplex import CPM_cplex from .pindakaas import CPM_pindakaas from .hexaly import CPM_hexaly +from .optalcp import CPM_optalcp from .rc2 import CPM_rc2 def param_combinations(all_params, remaining_keys=None, cur_params=None): @@ -90,6 +91,7 @@ def base_solvers(cls): ("cplex", CPM_cplex), ("pindakaas", CPM_pindakaas), ("hexaly", CPM_hexaly), + ("optalcp", CPM_optalcp), ("rc2", CPM_rc2), ("scip", CPM_scip), ] diff --git a/tests/test_solverinterface.py b/tests/test_solverinterface.py index 16f037a61..a1d6c89f7 100644 --- a/tests/test_solverinterface.py +++ b/tests/test_solverinterface.py @@ -4,9 +4,22 @@ from cpmpy.solvers.utils import SolverLookup import cpmpy as cp from cpmpy.expressions.utils import is_any_list -from cpmpy.exceptions import NotSupportedError +from cpmpy.exceptions import MaskedSolverValueError, NotSupportedError from utils import skip_on_missing_pblib + +def _assert_value_or_masked(solver_name, cpm_var, expected): + try: + value = cpm_var.value() + except MaskedSolverValueError as exc: + assert solver_name == "optalcp" + assert "Preview" in str(exc) + assert "Academic or Full" in str(exc) + return True + + assert value == expected + return False + @pytest.mark.usefixtures("solver") @skip_on_missing_pblib(skip_on_exception_only=True) def test_empty_constructor(solver): @@ -99,7 +112,11 @@ def test_solve(solver): assert solver.solve() assert solver.status().exitstatus == ExitStatus.FEASIBLE - assert [x.value(), y.value(), z.value()] == [0, 1, 0] + masked = any(_assert_value_or_masked(solver, var, exp) for var, exp in zip((x, y, z), (0, 1, 0))) + if masked: + for var in (x, y, z): + with pytest.raises(MaskedSolverValueError): + var.value() @pytest.mark.usefixtures("solver") From 3fcd0213c52b9e3d453607115f1d93404ee64e54 Mon Sep 17 00:00:00 2001 From: OrestisLomis Date: Mon, 18 May 2026 14:41:20 +0200 Subject: [PATCH 2/8] remove masked variable support --- cpmpy/expressions/variables.py | 10 ---------- cpmpy/model.py | 8 -------- cpmpy/solvers/optalcp.py | 9 +++------ tests/test_solverinterface.py | 20 ++------------------ 4 files changed, 5 insertions(+), 42 deletions(-) diff --git a/cpmpy/expressions/variables.py b/cpmpy/expressions/variables.py index 9f31862b0..80c93b36e 100644 --- a/cpmpy/expressions/variables.py +++ b/cpmpy/expressions/variables.py @@ -65,7 +65,6 @@ import numpy as np import cpmpy as cp # to avoid circular import -from ..exceptions import MaskedSolverValueError from .core import Expression, ExprLike, BoolExprLike, ListLike, BoolVal from .utils import is_num, is_int, is_boolexpr, get_bounds @@ -328,7 +327,6 @@ def __init__(self, lb: int, ub: int, name: str): self.ub = ub self.name = name self._value: Optional[int] = None - self._value_error: Optional[str] = None def has_subexpr(self) -> bool: """Does it contains nested Expressions? @@ -345,8 +343,6 @@ def value(self) -> Optional[int]: """ the value obtained in the last solve call (or 'None') """ - if self._value_error is not None: - raise MaskedSolverValueError(self._value_error) return self._value def get_bounds(self) -> tuple[int, int]: @@ -357,12 +353,6 @@ def clear(self) -> None: """ clear the value obtained from the last solve call """ self._value = None - self._value_error = None - - def _mask_value(self, message: str) -> None: - """Mark the last solver value as intentionally unavailable.""" - self._value = None - self._value_error = message def __repr__(self) -> str: return self.name diff --git a/cpmpy/model.py b/cpmpy/model.py index 127fe1c87..e07b7d371 100644 --- a/cpmpy/model.py +++ b/cpmpy/model.py @@ -57,7 +57,6 @@ def __init__(self, *args, minimize=None, maximize=None): """ assert ((minimize is None) or (maximize is None)), "can not set both minimize and maximize" self.cpm_status = SolverStatus("Model") # status of solving this model, will be replaced - self.objective_value_ = None # init list of constraints and objective self.constraints = [] @@ -148,7 +147,6 @@ def objective(self, expr, minimize): """ self.objective_ = expr self.objective_is_min = minimize - self.objective_value_ = None def has_objective(self): """ @@ -166,10 +164,6 @@ def objective_value(self): Returns: int, optional: The objective value as an integer or ``None`` if it is not run or is a satisfaction problem """ - if self.objective_ is None: - return None - if self.objective_value_ is not None: - return self.objective_value_ return self.objective_.value() def solve(self, solver:Optional[str]=None, time_limit:Optional[int|float]=None, **kwargs): @@ -203,7 +197,6 @@ def solve(self, solver:Optional[str]=None, time_limit:Optional[int|float]=None, ret = s.solve(time_limit=time_limit, **kwargs) # store CPMpy status (s object has no further use) self.cpm_status = s.status() - self.objective_value_ = s.objective_value() return ret def solveAll(self, solver:Optional[str]=None, display:Optional[Callback]=None, time_limit:Optional[int|float]=None, solution_limit:Optional[int]=None, **kwargs): @@ -234,7 +227,6 @@ def solveAll(self, solver:Optional[str]=None, display:Optional[Callback]=None, t ret = s.solveAll(display=display,time_limit=time_limit,solution_limit=solution_limit, call_from_model=True, **kwargs) # store CPMpy status (s object has no further use) self.cpm_status = s.status() - self.objective_value_ = s.objective_value() return ret def status(self): diff --git a/cpmpy/solvers/optalcp.py b/cpmpy/solvers/optalcp.py index 18585d195..47528ac7e 100644 --- a/cpmpy/solvers/optalcp.py +++ b/cpmpy/solvers/optalcp.py @@ -37,7 +37,7 @@ from .solver_interface import SolverInterface, SolverStatus, ExitStatus, Callback from .. import DirectConstraint -from ..exceptions import MaskedSolverValueError, NotSupportedError +from ..exceptions import NotSupportedError from ..expressions.core import Comparison, Operator, BoolVal from ..expressions.globalconstraints import GlobalConstraint from ..expressions.globalfunctions import GlobalFunction @@ -159,11 +159,8 @@ def solveAll(self, display: Optional[Callback] = None, time_limit: Optional[floa try: return super().solveAll(display=display, time_limit=time_limit, solution_limit=solution_limit, call_from_model=call_from_model, **kwargs) - except MaskedSolverValueError as exc: - raise NotSupportedError( - "OptalCP solveAll() requires visible decision-variable values. " - "The Preview edition masks them." - ) from exc + except Exception as exc: + raise exc def _populate_values(self, solution): if solution is None: diff --git a/tests/test_solverinterface.py b/tests/test_solverinterface.py index a1d6c89f7..46994af06 100644 --- a/tests/test_solverinterface.py +++ b/tests/test_solverinterface.py @@ -4,22 +4,10 @@ from cpmpy.solvers.utils import SolverLookup import cpmpy as cp from cpmpy.expressions.utils import is_any_list -from cpmpy.exceptions import MaskedSolverValueError, NotSupportedError +from cpmpy.exceptions import NotSupportedError from utils import skip_on_missing_pblib -def _assert_value_or_masked(solver_name, cpm_var, expected): - try: - value = cpm_var.value() - except MaskedSolverValueError as exc: - assert solver_name == "optalcp" - assert "Preview" in str(exc) - assert "Academic or Full" in str(exc) - return True - - assert value == expected - return False - @pytest.mark.usefixtures("solver") @skip_on_missing_pblib(skip_on_exception_only=True) def test_empty_constructor(solver): @@ -112,11 +100,7 @@ def test_solve(solver): assert solver.solve() assert solver.status().exitstatus == ExitStatus.FEASIBLE - masked = any(_assert_value_or_masked(solver, var, exp) for var, exp in zip((x, y, z), (0, 1, 0))) - if masked: - for var in (x, y, z): - with pytest.raises(MaskedSolverValueError): - var.value() + assert [x.value(), y.value(), z.value()] == [0, 1, 0] @pytest.mark.usefixtures("solver") From 480e0b4c5a6ca05beb9cdc46573471e5e7f389bf Mon Sep 17 00:00:00 2001 From: OrestisLomis Date: Tue, 19 May 2026 19:32:48 +0200 Subject: [PATCH 3/8] bugfixes --- cpmpy/expressions/utils.py | 10 +++++++++ cpmpy/solvers/optalcp.py | 45 +++++++++++++++++++------------------- 2 files changed, 33 insertions(+), 22 deletions(-) diff --git a/cpmpy/expressions/utils.py b/cpmpy/expressions/utils.py index 2d7867d95..302e0814a 100644 --- a/cpmpy/expressions/utils.py +++ b/cpmpy/expressions/utils.py @@ -61,6 +61,16 @@ def is_num(arg): """ return isinstance(arg, (bool, np.bool_, cp.BoolVal, int, np.integer, float, np.floating)) +def is_np_int(arg): + """ is it an numpy int? + """ + return isinstance(arg, np.integer) + +def is_np_bool(arg): + """ is it a numpy bool? + """ + return isinstance(arg, np.bool_) + def is_false_cst(arg): """ is the argument the constant False (can be of type bool, np.bool and BoolVal) diff --git a/cpmpy/solvers/optalcp.py b/cpmpy/solvers/optalcp.py index 47528ac7e..d571cb138 100644 --- a/cpmpy/solvers/optalcp.py +++ b/cpmpy/solvers/optalcp.py @@ -11,14 +11,13 @@ Requires the `optalcp` Python package and an OptalCP solver binary. - Public preview installation currently works via: + Using OptalCP through CPMpy requires a full or academic version, which can be installed through: .. code-block:: console - $ pip install git+https://github.com/ScheduleOpt/optalcp-py-bin-preview@latest + $ pip install git+https://github.com/ScheduleOpt/optalcp-py-bin-academic@latest - The preview edition solves models and reports objective values, but masks decision-variable assignments. - Academic and full editions expose variable values as well. + Visit https://dev.vilim.eu/docs/Quick%20Start/editions#installation-1 for more information on how to obtain non-preview versions of OptalCP. Documentation: https://optalcp.com/docs/ @@ -42,7 +41,7 @@ from ..expressions.globalconstraints import GlobalConstraint from ..expressions.globalfunctions import GlobalFunction from ..expressions.variables import _BoolVarImpl, NegBoolView, _IntVarImpl, _NumVarImpl, intvar -from ..expressions.utils import is_num, is_any_list, eval_comparison, get_bounds, get_nonneg_args, implies +from ..expressions.utils import is_num, is_np_int, is_np_bool, is_any_list, eval_comparison, get_bounds, get_nonneg_args, implies from ..transformations.get_variables import get_variables from ..transformations.normalize import toplevel_list from ..transformations.decompose_global import decompose_in_tree, decompose_objective @@ -64,11 +63,6 @@ class CPM_optalcp(SolverInterface): }) supported_reified_global_constraints = frozenset() - _masked_value_message = ( - "OptalCP Preview solved the model but masks decision-variable values. " - "Install an Academic or Full OptalCP edition to retrieve variable assignments." - ) - @staticmethod def installed(): try: @@ -107,7 +101,6 @@ def __init__(self, cpm_model=None, subsolver=None): assert subsolver is None self.opt_model = optalcp.Model() self._objective_is_min = None - self._values_masked = False super().__init__(name="optalcp", cpm_model=cpm_model) @property @@ -142,7 +135,6 @@ def solve(self, time_limit: Optional[float] = None, **kwargs): has_sol = self._solve_return(self.cpm_status) self.objective_value_ = None - self._values_masked = False if has_sol: self._populate_values(self.opt_result.solution) @@ -168,20 +160,11 @@ def _populate_values(self, solution): cpm_var.clear() return - masked = False values = {} for cpm_var in self.user_vars: opt_var = self.solver_var(cpm_var) value = solution.get_value(opt_var) values[cpm_var] = value - if value is None: - masked = True - - if masked: - self._values_masked = True - for cpm_var in self.user_vars: - cpm_var._mask_value(self._masked_value_message) - return for cpm_var, value in values.items(): cpm_var.clear() @@ -260,6 +243,12 @@ def _opt_expr(self, cpm_con): if isinstance(cpm_con, BoolVal): return cpm_con.args[0] + + if is_np_int(cpm_con): + return int(cpm_con) + + if is_np_bool(cpm_con): + return bool(cpm_con) if is_num(cpm_con): return cpm_con @@ -377,7 +366,19 @@ def _opt_cumulative(self, cpm_con): pulses.append(task.pulse(self._opt_expr(h))) if pulses: - cons.append(self.opt_model.sum(pulses) <= self._opt_expr(capacity)) + # OptalCP requires maxCapacity to be a "present" (non-optional) expression. + # If capacity is a complex expression (e.g. 3*bv[0]), reify it into a + # fresh auxiliary intvar that is guaranteed present. + if is_num(capacity) or isinstance(capacity, _IntVarImpl): + opt_cap = self._opt_expr(capacity) + else: + lb, ub = get_bounds(capacity) + cap_aux = intvar(lb, ub) + self.user_vars.add(cap_aux) + opt_cap = self.solver_var(cap_aux) + cons += self._opt_expr([cap_aux == capacity]) + + cons.append(self.opt_model.sum(pulses) <= opt_cap) return cons def _make_tasks(self, start, dur, end, is_present): From 7a32b750725286d87cd4abf5110a9772c96ff2aa Mon Sep 17 00:00:00 2001 From: OrestisLomis Date: Tue, 19 May 2026 20:56:13 +0200 Subject: [PATCH 4/8] add doc --- cpmpy/solvers/optalcp.py | 282 ++++++++++++++++++++++++++++++++++----- 1 file changed, 245 insertions(+), 37 deletions(-) diff --git a/cpmpy/solvers/optalcp.py b/cpmpy/solvers/optalcp.py index d571cb138..8d378167f 100644 --- a/cpmpy/solvers/optalcp.py +++ b/cpmpy/solvers/optalcp.py @@ -33,6 +33,7 @@ """ from typing import Optional +import warnings from .solver_interface import SolverInterface, SolverStatus, ExitStatus, Callback from .. import DirectConstraint @@ -55,6 +56,9 @@ class CPM_optalcp(SolverInterface): Creates the following attributes (see parent constructor for more): - ``opt_model``: object, OptalCP model object + + Documentation of the solver's own Python API: (all modeling functions) + https://dev.vilim.eu/python-api/index.html """ supported_global_constraints = frozenset({ @@ -65,25 +69,46 @@ class CPM_optalcp(SolverInterface): @staticmethod def installed(): + # try to import the package try: import optalcp return True except ModuleNotFoundError: return False - + @staticmethod - def supported(): + def license_ok(): if not CPM_optalcp.installed(): + warnings.warn( + "License check failed, python package 'optalcp' is not installed! " + "Please check 'CPM_optalcp.installed()' before attempting to check license." + ) return False try: import optalcp - optalcp.Solver.find_solver({}) - return True + mdl = optalcp.Model() + bv = mdl.bool_var(name="dummy") + mdl.enforce(bv) + result = mdl.solve({"printLog": False}) + if result.nb_solutions == 0: + return False + # Without a valid (non-preview) license, OptalCP returns masked/garbage values. + # A bool var enforced to True must be 1 if the license is valid. + return result.solution.get_value(bv) == 1 except Exception: return False + @staticmethod + def supported(): + return CPM_optalcp.installed() and CPM_optalcp.license_ok() + @staticmethod def version() -> Optional[str]: + """ + Returns the installed version of the solver's Python API. + + The version does not include whether the user has a preview version or not. + """ try: import optalcp return optalcp.__version__ @@ -91,10 +116,20 @@ def version() -> Optional[str]: return None def __init__(self, cpm_model=None, subsolver=None): + """ + Constructor of the native solver object + + Arguments: + cpm_model: Model(), a CPMpy Model() (optional) + """ if not self.installed(): raise ModuleNotFoundError( "CPM_optalcp: Install the 'optalcp' Python package and an OptalCP binary to use this solver interface." ) + + if not self.license_ok(): + raise ModuleNotFoundError("CPM_optalcp: In CPMpy we only support non-preview versions of OptalCP." \ + "You can request an academic/full version: https://dev.vilim.eu/docs/Quick%20Start/editions") import optalcp @@ -105,27 +140,64 @@ def __init__(self, cpm_model=None, subsolver=None): @property def native_model(self): + """ + Returns the solver's underlying native model (for direct solver access). + """ return self.opt_model def solve(self, time_limit: Optional[float] = None, **kwargs): + """ + Call the OptalCP solver + + Arguments: + time_limit (float, optional): maximum solve time in seconds + + kwargs: any keyword argument, sets parameters of solver object + + Arguments that correspond to solver parameters: + + + ============================= ============ + Argument Description + ============================= ============ + printLog This parameter controls the verbosity. It is a boolean. The default value is False. + absoluteGapTolerance This parameter sets an absolute tolerance on the objective value for optimization models. The value is a positive float. Default value is 0. + relativeGapTolerance This parameter sets a relative tolerance on the objective value for optimization models. The value is a positive float. Default value is 0.0001. This parameter works together with Parameters.relativeGapTolerance as an OR condition: the search stops when either the absolute gap or the relative gap is within tolerance. + seachType This parameter controls the type of search the solver uses. Possible values are: + - 'Auto': Automatically determined (Default) + - 'LNS': Large Neighbourhood Search + - 'FDS': Failure-Directed Search + - 'FDSDual': Failure-Directed Search working on objective bounds + - 'SetTimes': Depth-first set-times search + nbWorkers This parameter sets the number of workers to run in parallel to solve your model. The value is a positive integer. Default value is 0. (0 = use environment variable OPTALCP_NB_WORKERS or use all available CPU cores) + ============================= ============ + + All solver parameters are documented here: https://dev.vilim.eu/python-api/api.html#optalcp.Parameters + """ + # ensure all vars are known to solver self.solver_vars(list(self.user_vars)) + # edge case, empty model, ensure the solver has something to solve if not len(self.user_vars): self.add(intvar(1, 1) == 1) + # actual default value is True, but we do not want the log to be on by default in CPMpy if "printLog" not in kwargs: kwargs["printLog"] = False + # set time limit if time_limit is not None: if time_limit <= 0: raise ValueError("Time limit must be positive") kwargs["timeLimit"] = time_limit self.opt_result = self.opt_model.solve(kwargs or None) - + + # new status, translate runtime self.cpm_status = SolverStatus(self.name) - self.cpm_status.runtime = self.opt_result.duration + self.cpm_status.runtime = self.opt_result.duration # wallclock time in (float) seconds + # translate solver exit status to CPMpy exit status if self.opt_result.nb_solutions == 0: self.cpm_status.exitstatus = ExitStatus.UNSATISFIABLE if self.opt_result.proof else ExitStatus.UNKNOWN elif self.has_objective(): @@ -133,14 +205,24 @@ def solve(self, time_limit: Optional[float] = None, **kwargs): else: self.cpm_status.exitstatus = ExitStatus.FEASIBLE + # True/False depending on self.cpm_status has_sol = self._solve_return(self.cpm_status) - self.objective_value_ = None if has_sol: - self._populate_values(self.opt_result.solution) + # fill the variable values + for cpm_var in self.user_vars: + opt_var = self.solver_var(cpm_var) + value = self.opt_result.solution.get_value(opt_var) + if isinstance(cpm_var, _BoolVarImpl): + cpm_var._value = bool(value) + else: + cpm_var._value = value + + # translate objective, for optimisation problems only if self.has_objective(): self.objective_value_ = self.opt_result.objective - else: + + else: # clear values of variables for cpm_var in self.user_vars: cpm_var.clear() @@ -148,38 +230,43 @@ def solve(self, time_limit: Optional[float] = None, **kwargs): def solveAll(self, display: Optional[Callback] = None, time_limit: Optional[float] = None, solution_limit: Optional[int] = None, call_from_model=False, **kwargs): + """ + A shorthand to compute all (optimal) solutions, map them to CPMpy and optionally display the solutions. + + If the problem is an optimization problem, returns only optimal solutions. + + Arguments: + display: either a list of CPMpy expressions, OR a callback function, called with the variables after value-mapping. + Default is None, meaning nothing is displayed. + time_limit: Stop after this many seconds. Default is None. + solution_limit: Stop after this many solutions. Default is None. + call_from_model: Whether the method is called from a CPMpy Model instance or not. + **kwargs: Any other keyword arguments. + + Returns: + int: Number of solutions found. + """ try: return super().solveAll(display=display, time_limit=time_limit, solution_limit=solution_limit, call_from_model=call_from_model, **kwargs) except Exception as exc: raise exc - def _populate_values(self, solution): - if solution is None: - for cpm_var in self.user_vars: - cpm_var.clear() - return - - values = {} - for cpm_var in self.user_vars: - opt_var = self.solver_var(cpm_var) - value = solution.get_value(opt_var) - values[cpm_var] = value - - for cpm_var, value in values.items(): - cpm_var.clear() - if isinstance(cpm_var, _BoolVarImpl): - cpm_var._value = bool(value) - else: - cpm_var._value = value - + def solver_var(self, cpm_var): - if is_num(cpm_var): + """ + Creates solver variable for cpmpy variable + or returns from cache if previously created + """ + if is_num(cpm_var): # shortcut, eases posting constraints return cpm_var + # special case, negative-bool-view + # work directly on var inside the view if isinstance(cpm_var, NegBoolView): return ~self.solver_var(cpm_var._bv) + # create if it does not exist if cpm_var not in self._varmap: if isinstance(cpm_var, _BoolVarImpl): revar = self.opt_model.bool_var(name=str(cpm_var)) @@ -189,9 +276,20 @@ def solver_var(self, cpm_var): raise NotImplementedError(f"Not a known var {cpm_var}") self._varmap[cpm_var] = revar + # return from cache return self._varmap[cpm_var] def objective(self, expr, minimize=True): + """ + Post the given expression to the solver as objective to minimize/maximize + + ``objective()`` can be called multiple times, only the last one is stored + + .. note:: + + technical side note: any constraints created during conversion of the objective are permanently posted to the solver + """ + # save user variables get_variables(expr, self.user_vars) obj, decomp_cons = decompose_objective( @@ -212,7 +310,22 @@ def objective(self, expr, minimize=True): def has_objective(self): return self._objective_is_min is not None + # `add()` first calls `transform()` def transform(self, cpm_expr): + """ + Transform arbitrary CPMpy expressions to constraints the solver supports + + Implemented through chaining multiple solver-independent **transformation functions** from + the `cpmpy/transformations/` directory. + + See the :ref:`Adding a new solver` docs on readthedocs for more information. + + :param cpm_expr: CPMpy expression, or list thereof + :type cpm_expr: Expression or list of Expression + + :return: list of Expression + """ + # apply transformations cpm_cons = toplevel_list(cpm_expr) cpm_cons = no_partial_functions(cpm_cons, safen_toplevel=frozenset({})) cpm_cons = decompose_in_tree( @@ -221,12 +334,32 @@ def transform(self, cpm_expr): supported_reified=self.supported_reified_global_constraints, csemap=self._csemap ) + # no flattening required return cpm_cons def add(self, cpm_expr): + """ + Eagerly add a constraint to the underlying solver. + + Any CPMpy expression given is immediately transformed (through `transform()`) + and then posted to the solver in this function. + + This can raise 'NotImplementedError' for any constraint not supported after transformation + + The variables used in expressions given to add are stored as 'user variables'. Those are the only ones + the user knows and cares about (and will be populated with a value after solve). All other variables + are auxiliary variables created by transformations. + + :param cpm_expr: CPMpy expression, or list thereof + :type cpm_expr: Expression or list of Expression + + :return: self + """ + # add new user vars to the set get_variables(cpm_expr, collect=self.user_vars) for cpm_con in self.transform(cpm_expr): + # translate each expression tree, then post straight away opt_con = self._opt_expr(cpm_con) if is_any_list(opt_con): self.opt_model.enforce(opt_con) @@ -235,10 +368,17 @@ def add(self, cpm_expr): return self - __add__ = add + __add__ = add # avoid redirect in superclass def _opt_expr(self, cpm_con): + """ + OptalCP supports nested expressions, + so we recursively translate our expressions to theirs. + + Accepts a single constraint or a list thereof; return type changes accordingly. + """ if is_any_list(cpm_con): + # arguments can be lists return [self._opt_expr(con) for con in cpm_con] if isinstance(cpm_con, BoolVal): @@ -255,7 +395,8 @@ def _opt_expr(self, cpm_con): if isinstance(cpm_con, _NumVarImpl): return self.solver_var(cpm_con) - + + # Operators: base (bool), lhs=numexpr, lhs|rhs=boolexpr (reified ->) if isinstance(cpm_con, Operator): if cpm_con.name == "and": args = self._opt_expr(cpm_con.args) @@ -291,10 +432,12 @@ def _opt_expr(self, cpm_con): return -self._opt_expr(cpm_con.args[0]) raise NotImplementedError(f"Operator {cpm_con} not implemented for OptalCP") + # Comparisons (just translate the subexpressions and re-post) if isinstance(cpm_con, Comparison): lhs, rhs = self._opt_expr(cpm_con.args) + # post the comparison return eval_comparison(cpm_con.name, lhs, rhs) - + # rest: base (Boolean) global constraints if isinstance(cpm_con, GlobalConstraint): if cpm_con.name in ("cumulative", "cumulative_optional"): return self._opt_cumulative(cpm_con) @@ -342,6 +485,18 @@ def _opt_expr(self, cpm_con): raise NotImplementedError("OptalCP: constraint not (yet) supported", cpm_con) def _opt_cumulative(self, cpm_con): + """ + Helper function to translate CPMpy's Cumulative and CumulativeOptional constraints + to OptalCP's pulse-based cumulative expressions. + + Creates interval variables for each task via `_make_tasks`, then posts + the sum of pulses <= capacity constraint. + + OptalCP requires the capacity expression to be 'present' (non-optional). + If capacity is a complex expression (e.g. involving boolean variables), + it is reified into a fresh auxiliary intvar to satisfy this requirement. + """ + # unpack arguments depending on constraint type and whether end times are provided if cpm_con.name == "cumulative": is_present = None if len(cpm_con.args) == 4: @@ -349,17 +504,23 @@ def _opt_cumulative(self, cpm_con): end = None else: start, dur, end, demand, capacity = cpm_con.args - else: + else: # cumulative_optional if len(cpm_con.args) == 5: start, dur, demand, capacity, is_present = cpm_con.args end = None else: start, dur, end, demand, capacity, is_present = cpm_con.args + # create interval variables and linking constraints for each task tasks, cons = self._make_tasks(start, dur, end, is_present) + + # get_nonneg_args strips the presence multiplication from demand expressions + # (e.g. bv * [2,3,4] -> [2,3,4]) since the optional interval var already + # contributes 0 pulse when absent — no need to encode it in the demand demand_lst, demand_cons = get_nonneg_args(demand, is_present) cons += self._opt_expr(demand_cons) + # build a pulse for each present (non-zero-duration) task pulses = [] for task, h in zip(tasks, demand_lst): if task is not None: @@ -382,14 +543,31 @@ def _opt_cumulative(self, cpm_con): return cons def _make_tasks(self, start, dur, end, is_present): + """ + Helper function to create a list of OptalCP interval variables and additional + constraints enforcing the relationship between CPMpy start/duration/end variables + and the native interval variables. + + Arguments: + start: list of CPMpy integer expressions for task start times + dur: list of CPMpy integer expressions for task durations + end: list of CPMpy integer expressions for task end times, or None + is_present: list of CPMpy boolean variables indicating task presence, or None + + Returns: + tasks: list of OptalCP interval variables (None for zero-duration tasks) + cons: list of OptalCP constraints linking the interval variables to + the CPMpy start/duration/end/presence variables + """ if end is None: - end = [None for _ in range(len(start))] + end = [None for _ in range(len(start))] # easier to handle the task-making below + # OptalCP crashes if size of interval is negative dur, dur_cons = get_nonneg_args(dur, is_present) extra_cons = self._opt_expr(dur_cons) if is_present is None: - is_present = [None] * len(start) + is_present = [None] * len(start) # eases handling below tasks = [] for s, d, e, p in zip(start, dur, end, is_present): @@ -399,19 +577,45 @@ def _make_tasks(self, start, dur, end, is_present): return tasks, extra_cons def _make_task(self, start, dur, end, is_present): + """ + Helper function to create a single OptalCP interval variable and additional + constraints enforcing the relationship between CPMpy start/duration/end variables + and the native interval variable. + + Zero-duration tasks (lb == ub == 0) are handled as a special case: no interval + variable is created; instead, a constraint start == end is posted (if end is given). + + For optional tasks (is_present is not None), the interval variable is created + with optional=True, and its presence is tied to the CPMpy boolean variable + via a presence() == is_present constraint. Start, length, and end are linked + through implication constraints conditioned on presence. + + Arguments: + start: CPMpy integer expression for the task start time + dur: CPMpy integer expression for the task duration (must be non-negative) + end: CPMpy integer expression for the task end time, or None + is_present: CPMpy boolean variable indicating task presence, or None + + Returns: + task: OptalCP interval variable, or None for zero-duration tasks + cons: list of OptalCP constraints linking the interval variable to + the CPMpy start/duration/end/presence variables + """ assert get_bounds(dur)[0] >= 0, "optalcp does not support intervals with negative duration, use `utils.get_nonneg_args` first" is_optional = is_present is not None if not is_optional: - is_present = BoolVal(True) + is_present = BoolVal(True) # eases handling below lb, ub = get_bounds(dur) extra_cons = [] if lb == 0 == ub: - if end is None: + if end is None: # nothing to enforce return None, [] return None, self._opt_expr([implies(is_present, start == end)]) + # create the interval variable with domain bounds derived from the CPMpy expressions; + # mark it optional when a presence variable was provided task = self.opt_model.interval_var( start=get_bounds(start), end=get_bounds(end) if end is not None else None, @@ -420,12 +624,16 @@ def _make_task(self, start, dur, end, is_present): ) if is_optional: + # tie the interval's presence status to the CPMpy boolean variable extra_cons.append(task.presence() == self.solver_var(is_present)) + # when present, link start/length/end to the CPMpy expressions via implications + # so that the solver correctly propagates values from interval to CPMpy variables extra_cons.append(self.opt_model.implies(task.presence(), task.start() == self._opt_expr(start))) extra_cons.append(self.opt_model.implies(task.presence(), task.length() == self._opt_expr(dur))) if end is not None: extra_cons.append(self.opt_model.implies(task.presence(), task.end() == self._opt_expr(end))) else: + # mandatory task: link unconditionally extra_cons.append(task.start() == self._opt_expr(start)) extra_cons.append(task.length() == self._opt_expr(dur)) if end is not None: From b07e3736a501129a32162578b2bbbdb36481683c Mon Sep 17 00:00:00 2001 From: OrestisLomis Date: Tue, 19 May 2026 20:57:49 +0200 Subject: [PATCH 5/8] remove masked exception --- cpmpy/exceptions.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/cpmpy/exceptions.py b/cpmpy/exceptions.py index 0cf9f4732..fa316c58e 100644 --- a/cpmpy/exceptions.py +++ b/cpmpy/exceptions.py @@ -43,8 +43,3 @@ class GCSVerificationException(CPMpyException): class TransformationNotImplementedError(CPMpyException): '''Raised when a transformation is not implemented for a certain expression''' pass - - -class MaskedSolverValueError(CPMpyException): - '''Raised when a solver found a solution but intentionally masked decision-variable values''' - pass From 9e9aef5b8c2d7270d30127e786788641d448c09d Mon Sep 17 00:00:00 2001 From: OrestisLomis Date: Wed, 20 May 2026 11:31:12 +0200 Subject: [PATCH 6/8] update --- cpmpy/solvers/optalcp.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/cpmpy/solvers/optalcp.py b/cpmpy/solvers/optalcp.py index 8d378167f..e4b2b4e66 100644 --- a/cpmpy/solvers/optalcp.py +++ b/cpmpy/solvers/optalcp.py @@ -1,3 +1,8 @@ +#!/usr/bin/env python +#-*- coding:utf-8 -*- +## +## optalcp.py +## """ Interface to OptalCP's Python API. From 0178073f91046d948cc1b20297080ed7d4cd11b2 Mon Sep 17 00:00:00 2001 From: OrestisLomis Date: Wed, 20 May 2026 11:34:10 +0200 Subject: [PATCH 7/8] update --- cpmpy/solvers/optalcp.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cpmpy/solvers/optalcp.py b/cpmpy/solvers/optalcp.py index e4b2b4e66..a46251142 100644 --- a/cpmpy/solvers/optalcp.py +++ b/cpmpy/solvers/optalcp.py @@ -76,7 +76,7 @@ class CPM_optalcp(SolverInterface): def installed(): # try to import the package try: - import optalcp + import optalcp # type: ignore[import-not-found] return True except ModuleNotFoundError: return False From 8bfa4a19e367878b0f73f13977d1441f6e32e2d9 Mon Sep 17 00:00:00 2001 From: OrestisLomis Date: Wed, 27 May 2026 17:24:22 +0200 Subject: [PATCH 8/8] rewrite simple guarded demand to use optional interval --- cpmpy/solvers/optalcp.py | 89 +++++++++++++++++++++++++++++++++++++--- 1 file changed, 84 insertions(+), 5 deletions(-) diff --git a/cpmpy/solvers/optalcp.py b/cpmpy/solvers/optalcp.py index a46251142..67024f349 100644 --- a/cpmpy/solvers/optalcp.py +++ b/cpmpy/solvers/optalcp.py @@ -54,6 +54,71 @@ from ..transformations.safening import no_partial_functions +def _same_expression(lhs, rhs): + """Return whether two CPMpy expressions refer to the same syntactic expression.""" + return lhs is rhs or (type(lhs) is type(rhs) and repr(lhs) == repr(rhs)) + + +def _strip_presence_factor(expr, presence): + """ + Strip `presence` from simple demand expressions like `presence * height`. + + Returns the stripped expression when `expr` is a direct multiplication by + `presence`, otherwise returns `expr` unchanged. + """ + if _same_expression(expr, presence): + return 1 + + if getattr(expr, "name", None) != "mul" or len(expr.args) != 2: + return expr + + lhs, rhs = expr.args + if _same_expression(lhs, presence): + return rhs + if _same_expression(rhs, presence): + return lhs + return expr + + +def _extract_guarded_demands(demand): + """ + Detect simple `presence * height` cumulative demands. + + Returns `(is_present, stripped_demand)`. If no guarded demand is found, + `is_present` is `None`. + """ + is_present = [] + stripped_demand = [] + found_guard = False + + for expr in demand: + if isinstance(expr, _BoolVarImpl): + is_present.append(expr) + stripped_demand.append(1) + found_guard = True + continue + + if getattr(expr, "name", None) == "mul" and len(expr.args) == 2: + lhs, rhs = expr.args + if isinstance(lhs, _BoolVarImpl): + is_present.append(lhs) + stripped_demand.append(rhs) + found_guard = True + continue + if isinstance(rhs, _BoolVarImpl): + is_present.append(rhs) + stripped_demand.append(lhs) + found_guard = True + continue + + is_present.append(BoolVal(True)) + stripped_demand.append(expr) + + if not found_guard: + return None, demand + return is_present, stripped_demand + + class CPM_optalcp(SolverInterface): """ Interface to OptalCP's Python API. @@ -516,12 +581,26 @@ def _opt_cumulative(self, cpm_con): else: start, dur, end, demand, capacity, is_present = cpm_con.args + if is_present is None: + is_present, demand = _extract_guarded_demands(demand) + cons = [] + if is_present is not None: + # `Cumulative` enforces task consistency unconditionally, even + # when a guarded demand lets us model the pulse as optional. + _, dur_cons = get_nonneg_args(dur) + cons += self._opt_expr(dur_cons) + if end is not None: + cons += self._opt_expr([s + d == e for s, d, e in zip(start, dur, end)]) + else: + demand = [_strip_presence_factor(h, p) for h, p in zip(demand, is_present)] + cons = [] + # create interval variables and linking constraints for each task - tasks, cons = self._make_tasks(start, dur, end, is_present) + tasks, task_cons = self._make_tasks(start, dur, end, is_present) + cons += task_cons - # get_nonneg_args strips the presence multiplication from demand expressions - # (e.g. bv * [2,3,4] -> [2,3,4]) since the optional interval var already - # contributes 0 pulse when absent — no need to encode it in the demand + # Presence multipliers that match the optional interval have been stripped + # above; the optional interval already contributes 0 pulse when absent. demand_lst, demand_cons = get_nonneg_args(demand, is_present) cons += self._opt_expr(demand_cons) @@ -630,7 +709,7 @@ def _make_task(self, start, dur, end, is_present): if is_optional: # tie the interval's presence status to the CPMpy boolean variable - extra_cons.append(task.presence() == self.solver_var(is_present)) + extra_cons.append(task.presence() == self._opt_expr(is_present)) # when present, link start/length/end to the CPMpy expressions via implications # so that the solver correctly propagates values from interval to CPMpy variables extra_cons.append(self.opt_model.implies(task.presence(), task.start() == self._opt_expr(start)))