Scheduling with optional tasks

examples/rcpsp_alternative/run_cpsat.py

@@ -0,0 +1,68 @@
+#  Copyright (c) 2026 AIRBUS and its affiliates.
+#  This source code is licensed under the MIT license found in the
+#  LICENSE file in the root directory of this source tree.
+
+import logging
+
+from discrete_optimization.generic_tools.cp_tools import ParametersCp
+from discrete_optimization.rcpsp.parser import get_data_available, parse_file
+from discrete_optimization.rcpsp.utils import plot_ressource_view, plot_task_gantt, plt
+from discrete_optimization.rcpsp_alternative.problem import get_optional_tasks_done
+from discrete_optimization.rcpsp_alternative.solvers.cpsat import (
+    CpsatRcpspWithAlternativePathSolver,
+)
+from discrete_optimization.rcpsp_alternative.solvers.cpsat_auto import (
+    CpsatAutoRcpspWithAlternativePathSolver,
+)
+from discrete_optimization.rcpsp_alternative.utils import create_problem_rcpsp
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+def run_cpsat():
+    problem = parse_file([f for f in get_data_available() if "j601_1.sm" in f][0])
+    problem = create_problem_rcpsp(
+        problem,
+        nb_alternative_paths=5,
+        range_nb_subpath=(1, 4),
+        range_len_subpath=(3, 5),
+    )
+    solver = CpsatRcpspWithAlternativePathSolver(problem)
+    solver.init_model(strict_alternative_path=True)
+    res = solver.solve(parameters_cp=ParametersCp.default_cpsat(), time_limit=30)
+    sol = res[-1][0]
+    print(problem.evaluate(sol), problem.satisfy(sol))
+    print(get_optional_tasks_done(sol, problem))
+    plot_task_gantt(problem, sol)
+    plot_ressource_view(problem, sol)
+    plt.show()
+
+
+def run_cpsat_auto():
+    problem = parse_file([f for f in get_data_available() if "j601_1.sm" in f][0])
+    # problem = parse_file([f for f in get_data_available() if "j1010_1.mm" in f][0])
+
+    problem = create_problem_rcpsp(
+        problem,
+        nb_alternative_paths=5,
+        range_nb_subpath=(1, 4),
+        range_len_subpath=(3, 5),
+    )
+    solver = CpsatAutoRcpspWithAlternativePathSolver(problem)
+    solver.init_model(use_cpm_for_task_bounds=False, use_energy_constraints=False)
+    res = solver.solve(
+        parameters_cp=ParametersCp.default_cpsat(),
+        time_limit=30,
+        ortools_cpsat_solver_kwargs={"log_search_progress": True},
+    )
+    sol = res[-1][0]
+    print(problem.evaluate(sol), problem.satisfy(sol))
+    print("Optional tasks done : ", get_optional_tasks_done(sol, problem))
+    plot_task_gantt(problem, sol)
+    plot_ressource_view(problem, sol)
+    plt.show()
+
+
+if __name__ == "__main__":
+    run_cpsat_auto()

src/discrete_optimization/alb/rcalbp/problem.py

@@ -62,6 +62,7 @@
     BaseALBSolution,
     ResourceTaskData,
 )
+from discrete_optimization.generic_tasks_tools import AbsentValue
 from discrete_optimization.generic_tasks_tools.allocation import (
     UnaryResource,
 )
@@ -107,6 +108,12 @@ class RCALBPSolution(
 
     problem: "RCALBPProblem"
 
+    def is_present(self, task: Task) -> bool:
+        return task in self.task_assignment and self.task_assignment[task] not in {
+            None,
+            AbsentValue.ABSENT,
+        }
+
     def get_renewable_resource_consumption(self, resource: Resource, task: Task) -> int:
         return self.problem.get_task_demand(task, resource)
 
@@ -454,6 +461,9 @@ class RCALBPProblem(
     def renewable_resources_list(self) -> list[Resource]:
         return list(set(self.resources) | self.shared_resources)
 
+    def is_optional(self, task: Task) -> bool:
+        return False
+
     def get_resource_availabilities(
         self, resource: Resource
     ) -> list[tuple[int, int, int]]:

src/discrete_optimization/alb/rcalbp_l/problem.py

@@ -10,11 +10,13 @@
 from matplotlib import pyplot as plt
 from matplotlib.widgets import Slider
 
+from discrete_optimization.generic_tasks_tools import AbsentValue
 from discrete_optimization.generic_tasks_tools.allocation import (
     AllocationProblem,
     AllocationSolution,
     UnaryResource,
 )
+from discrete_optimization.generic_tasks_tools.base import NoOptionalTasksProblem
 from discrete_optimization.generic_tasks_tools.scheduling import (
     SchedulingProblem,
     SchedulingSolution,
@@ -66,6 +68,12 @@ def __init__(
         self.ramp_up_duration = ramp_up_duration
         self.nb_adjustments = nb_adjustments
 
+    def is_present(self, task: Task) -> bool:
+        return task[0] in self.wks and self.wks[task[0]] not in {
+            None,
+            AbsentValue.ABSENT,
+        }
+
     def is_allocated(self, task: Task, unary_resource: WorkStation) -> bool:
         return self.wks[task[0]] == unary_resource
 
@@ -140,7 +148,11 @@ def __init__(
         self.raw = sol.raw
 
 
-class RCALBPLProblem(SchedulingProblem[Task], AllocationProblem[Task, WorkStation]):
+class RCALBPLProblem(
+    SchedulingProblem[Task],
+    AllocationProblem[Task, WorkStation],
+    NoOptionalTasksProblem[Task],
+):
     """
     Problem definition for Resource-Constrained Assembly Line Balancing
     with Learning Effect (RC-ALBP/L).

src/discrete_optimization/alb/salbp/problem.py

@@ -10,9 +10,11 @@
     BaseALBSolution,
     TaskData,
 )
+from discrete_optimization.generic_tasks_tools import AbsentValue
 from discrete_optimization.generic_tasks_tools.allocation import (
     UnaryResource,
 )
+from discrete_optimization.generic_tasks_tools.base import NoOptionalTasksProblem
 from discrete_optimization.generic_tools.do_problem import (
     EncodingRegister,
     ModeOptim,
@@ -47,6 +49,9 @@ def __init__(self, problem: "SalbpProblem", allocation_to_station: list[int]):
         self._nb_stations = len(set(self.allocation_to_station))
         self._cached_schedule = None  # Cache for greedy schedule
 
+    def is_present(self, task: Task) -> bool:
+        return self.allocation_to_station[task] not in {None, AbsentValue.ABSENT}
+
     # BaseALBSolution interface implementation
     def get_station_index(self, task: Task) -> int:
         """Get the index of the station where task is assigned."""
@@ -138,7 +143,7 @@ def __eq__(self, other):
         return self.allocation_to_station == other.allocation_to_station
 
 
-class SalbpProblem(BaseALBProblem[int, int]):
+class SalbpProblem(BaseALBProblem[int, int], NoOptionalTasksProblem[Task]):
     """
     Simple Assembly Line Balancing Problem.
 

src/discrete_optimization/binpack/problem.py

@@ -9,10 +9,12 @@
 from dataclasses import dataclass, field
 from typing import Hashable
 
+from discrete_optimization.generic_tasks_tools import AbsentValue
 from discrete_optimization.generic_tasks_tools.allocation import (
     AllocationProblem,
     AllocationSolution,
 )
+from discrete_optimization.generic_tasks_tools.base import NoOptionalTasksProblem, Task
 from discrete_optimization.generic_tasks_tools.scheduling import (
     SchedulingProblem,
     SchedulingSolution,
@@ -49,6 +51,12 @@ def copy(self) -> BinPackSolution:
             problem=self.problem, allocation=deepcopy(self.allocation)
         )
 
+    def is_present(self, task: Task) -> bool:
+        return (
+            self.allocation[task] is not None
+            and self.allocation[task] != AbsentValue.ABSENT
+        )
+
     def get_end_time(self, task: Item) -> int:
         return self.allocation[task] + 1
 
@@ -81,7 +89,11 @@ class BinInstance:
     compatible_items: set[int] | None = field(default=None)
 
 
-class BinPackProblemBinType(AllocationProblem[Item, BinPack], SchedulingProblem[Item]):
+class BinPackProblemBinType(
+    AllocationProblem[Item, BinPack],
+    SchedulingProblem[Item],
+    NoOptionalTasksProblem[Task],
+):
     def __init__(
         self,
         list_items: list[ItemBinPack],

src/discrete_optimization/coloring/problem.py

@@ -15,10 +15,12 @@
 
 import numpy as np
 
+from discrete_optimization.generic_tasks_tools import AbsentValue
 from discrete_optimization.generic_tasks_tools.allocation import (
     AllocationProblem,
     AllocationSolution,
 )
+from discrete_optimization.generic_tasks_tools.base import NoOptionalTasksProblem, Task
 from discrete_optimization.generic_tools.do_problem import (
     ModeOptim,
     ObjectiveDoc,
@@ -71,6 +73,9 @@ def __init__(
         self.nb_color = nb_color
         self.nb_violations = nb_violations
 
+    def is_present(self, task: Task) -> bool:
+        return self.colors[task] is not None and self.colors[task] != AbsentValue.ABSENT
+
     def copy(self) -> ColoringSolution:
         """Efficient way of copying a coloring solution without deepcopying unnecessary attribute (problem).
 
@@ -239,7 +244,7 @@ def nodes_fixed(self) -> set[Hashable]:
             return set()
 
 
-class ColoringProblem(AllocationProblem[Node, Color]):
+class ColoringProblem(AllocationProblem[Node, Color], NoOptionalTasksProblem[Node]):
     """Coloring problem class implementation.
 
     Attributes:
@@ -277,6 +282,9 @@ def __init__(
         self.constraints_coloring = constraints_coloring
         self.has_constraints_coloring = constraints_coloring is not None
 
+    def is_optional(self, task: Task) -> bool:
+        return False
+
     @property
     def tasks_list(self) -> list[Node]:
         return self.nodes_name

src/discrete_optimization/facility/problem.py

@@ -22,6 +22,7 @@
     AllocationProblem,
     AllocationSolution,
 )
+from discrete_optimization.generic_tasks_tools.base import Task
 from discrete_optimization.generic_tools.do_problem import (
     ModeOptim,
     ObjectiveDoc,
@@ -103,6 +104,9 @@ def change_problem(self, new_problem: Problem) -> None:
     def is_allocated(self, task: Customer, unary_resource: Facility) -> bool:
         return self.facility_for_customers[task.index] == unary_resource.index
 
+    def is_present(self, task: Task) -> bool:
+        return self.facility_for_customers[task] is not None
+
 
 class FacilityProblem(AllocationProblem[Customer, Facility]):
     """Base class for the facility problem.
@@ -128,6 +132,9 @@ def __init__(
         self.facilities = facilities
         self.customers = customers
 
+    def is_optional(self, task: Task) -> bool:
+        return False
+
     @property
     def unary_resources_list(self) -> list[Facility]:
         return self.facilities

src/discrete_optimization/flex_scheduling/problem.py

@@ -2,7 +2,7 @@
 #  This source code is licensed under the MIT license found in the
 #  LICENSE file in the root directory of this source tree.
 from copy import deepcopy
-from dataclasses import dataclass, field
+from dataclasses import field
 from functools import cache
 from typing import Dict, Hashable, List, Set, Tuple, Type
 
@@ -308,6 +308,9 @@ def __init__(
         self.schedule = schedule
         self.modes = modes
 
+    def is_present(self, task: Task) -> bool:
+        return self.modes[self.problem.task_id_to_index[task]] is not None
+
     def get_mode(self, task: Task) -> int:
         index = self.problem.task_id_to_index[task]
         return self.modes[index]
@@ -338,6 +341,9 @@ class FlexProblem(
     ],
     WithoutAllocationProblem[Task],
 ):
+    def is_optional(self, task: Task) -> bool:
+        return False
+
     @property
     def non_skill_cumulative_resources_list(self) -> list[Skill]:
         return [resource.id for resource in self.resources if resource.renewable]

src/discrete_optimization/generic_tasks_tools/__init__.py

@@ -0,0 +1,8 @@
+#  Copyright (c) 2026 AIRBUS and its affiliates.
+#  This source code is licensed under the MIT license found in the
+#  LICENSE file in the root directory of this source tree.
+from enum import Enum
+
+
+class AbsentValue(Enum):
+    ABSENT = "absent"