Source code for discrete_optimization.generic_tasks_tools.transformations.generic_scheduling_impl
# 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 abc import abstractmethod
from typing import Generic, Optional, TypeVar
from discrete_optimization.generic_tasks_tools.enums import MinOrMax, StartOrEnd
from discrete_optimization.generic_tasks_tools.generic_scheduling import (
GenericSchedulingProblem,
GenericSchedulingSolution,
)
from discrete_optimization.generic_tasks_tools.generic_scheduling_impl import (
AvailabilityIntervals,
CumulativeResource,
GenericSchedulingImplProblem,
GenericSchedulingImplSolution,
NonRenewableResource,
NonSkillCumulativeResource,
Skill,
Task,
UnaryAvailabilityIntervals,
UnaryResource,
)
from discrete_optimization.generic_tasks_tools.generic_scheduling_utils import (
RawSolution,
TaskVariable,
)
from discrete_optimization.generic_tasks_tools.objectives.objective_computer import (
ObjectiveComputer,
)
from discrete_optimization.generic_tools.transformation import ProblemTransformation
SpecificSchedulingProblem = TypeVar(
"SpecificSchedulingProblem", bound=GenericSchedulingProblem
)
SpecificSchedulingSolution = TypeVar(
"SpecificSchedulingSolution", bound=GenericSchedulingSolution
)
[docs]
class ToGenericSchedulingImpl(
ProblemTransformation[
SpecificSchedulingProblem,
SpecificSchedulingSolution,
GenericSchedulingImplProblem,
GenericSchedulingImplSolution,
],
Generic[
SpecificSchedulingProblem,
SpecificSchedulingSolution,
],
):
"""Transform a specific scheduling problem into the generic implementation.
This is still an abstract class, as `convert_solution_from_raw_generic_to_specific()` remains to implement.
"""
def __init__(self, list_objective_computer: list[ObjectiveComputer] = None):
self._list_objective_computer = list_objective_computer
[docs]
@abstractmethod
def transform_solution_from_raw_generic_to_specific(
self,
raw_sol: RawSolution[Task, UnaryResource, Skill],
source_problem: SpecificSchedulingProblem,
) -> SpecificSchedulingSolution:
"""Convert a raw solution (from generic problem) into a specific solution to the source problem.
Args:
raw_sol: Raw solution to convert to a specific solution
source_problem: Target Problem we want to build a solution for.
Returns:
"""
...
[docs]
def back_transform_solution(
self,
solution: GenericSchedulingImplSolution,
source_problem: SpecificSchedulingProblem,
) -> SpecificSchedulingSolution:
return self.transform_solution_from_raw_generic_to_specific(
raw_sol=solution.raw_sol, source_problem=source_problem
)
[docs]
def get_list_objective_computer(
self,
source_problem: SpecificSchedulingProblem,
) -> list[ObjectiveComputer]:
"""From the source problem, compute a list of objective computers to be used in `GenericSchedulingImplProblem`.
To be overriden in child classes.
Returns:
"""
# We put None as problem, it will be linked to its parent scheduling problem, when calling __init__
# of GenericSchedulingImplProblem
if self._list_objective_computer is not None:
return self._list_objective_computer
return source_problem.get_list_objective_computer()
[docs]
def transform_problem(
self,
source_problem: SpecificSchedulingProblem,
) -> GenericSchedulingImplProblem:
horizon = source_problem.get_makespan_upper_bound()
durations_per_mode: dict[Task, dict[int, int]] = {
task: {
mode: source_problem.get_task_mode_duration(task=task, mode=mode)
for mode in source_problem.get_task_modes(task=task)
}
for task in source_problem.tasks_list
}
(
resources_consumption,
resources_consumption_dependent,
) = _extract_resource_consumptions(source_problem)
successors: dict[Task, set[Task]] = {
k: set(v) for k, v in source_problem.get_precedence_constraints().items()
}
unary_resources: set[UnaryResource] = set(source_problem.unary_resources_list)
unary_resources_skills: dict[UnaryResource, dict[Skill, int]] = {
unary_resource: {
skill: value
for skill in source_problem.skills_list
if (
value := source_problem.get_unary_resource_skill_value(
unary_resource=unary_resource, skill=skill
)
)
> 0
}
for unary_resource in source_problem.unary_resources_list
}
unary_resources_availabilities: dict[
UnaryResource, UnaryAvailabilityIntervals
] = {
unary_resource: [
(start, end)
for start, end, value in source_problem.get_resource_availabilities(
resource=unary_resource
)
if value > 0
]
for unary_resource in source_problem.unary_resources_list
}
unary_resources_task_compatibility: dict[Task, set[UnaryResource]] = {
task: {
unary_resource
for unary_resource in source_problem.unary_resources_list
if source_problem.is_compatible_task_unary_resource(
task=task, unary_resource=unary_resource
)
}
for task in source_problem.tasks_list
}
skills: set[Skill] = set(source_problem.skills_list)
non_skill_cumulative_resources: dict[
NonSkillCumulativeResource, int | AvailabilityIntervals
] = {
resource: source_problem.get_resource_availabilities(resource=resource)
for resource in source_problem.non_skill_cumulative_resources_list
}
non_renewable_resources: dict[NonRenewableResource, int] = {
resource: source_problem.get_non_renewable_resource_capacity(
resource=resource
)
for resource in source_problem.non_renewable_resources_list
}
time_windows: dict[
Task, tuple[int | None, int | None, int | None, int | None]
] = {
task: (
source_problem.get_task_start_or_end_lower_bound(
task=task, start_or_end=StartOrEnd.START
),
source_problem.get_task_start_or_end_lower_bound(
task=task, start_or_end=StartOrEnd.END
),
source_problem.get_task_start_or_end_upper_bound(
task=task, start_or_end=StartOrEnd.START
),
source_problem.get_task_start_or_end_upper_bound(
task=task, start_or_end=StartOrEnd.END
),
)
for task in source_problem.tasks_list
}
start_to_start_min_time_lags: list[tuple[Task, Task, int]] = (
_construct_min_only_time_lags_from_min_and_max_time_lags(
source_problem=source_problem,
task1_start_or_end=StartOrEnd.START,
task2_start_or_end=StartOrEnd.START,
)
)
start_to_end_min_time_lags: list[tuple[Task, Task, int]] = (
_construct_min_only_time_lags_from_min_and_max_time_lags(
source_problem=source_problem,
task1_start_or_end=StartOrEnd.START,
task2_start_or_end=StartOrEnd.END,
)
)
end_to_start_min_time_lags: list[tuple[Task, Task, int]] = (
_construct_min_only_time_lags_from_min_and_max_time_lags(
source_problem=source_problem,
task1_start_or_end=StartOrEnd.END,
task2_start_or_end=StartOrEnd.START,
)
)
end_to_end_min_time_lags: list[tuple[Task, Task, int]] = (
_construct_min_only_time_lags_from_min_and_max_time_lags(
source_problem=source_problem,
task1_start_or_end=StartOrEnd.END,
task2_start_or_end=StartOrEnd.END,
)
)
no_overlap_sets: set[frozenset[Task]] = source_problem.get_no_overlap()
forbidden_intervals: dict[Task, list[tuple[int, int]]] = {
task: source_problem.get_forbidden_intervals(task)
for task in source_problem.tasks_list
}
optional_tasks = set(source_problem.optional_tasks_list)
list_objective_computer = self.get_list_objective_computer(source_problem)
return GenericSchedulingImplProblem(
horizon=horizon,
durations_per_mode=durations_per_mode,
resource_consumptions=resources_consumption,
resource_consumptions_dependent=resources_consumption_dependent,
successors=successors,
unary_resources=unary_resources,
unary_resources_skills=unary_resources_skills,
unary_resources_availabilities=unary_resources_availabilities,
unary_resources_task_compatibility=unary_resources_task_compatibility,
skills=skills,
non_skill_cumulative_resources=non_skill_cumulative_resources,
non_renewable_resources=non_renewable_resources,
time_windows=time_windows,
start_to_start_min_time_lags=start_to_start_min_time_lags,
start_to_end_min_time_lags=start_to_end_min_time_lags,
end_to_start_min_time_lags=end_to_start_min_time_lags,
end_to_end_min_time_lags=end_to_end_min_time_lags,
no_overlap_sets=no_overlap_sets,
forbidden_intervals=forbidden_intervals,
flexible_gap_blocking_constraints=source_problem.get_flexible_gap_blocking_constraints(),
span_blocking_constraints=source_problem.get_span_blocking_constraints(),
mode_constraints=source_problem.get_mode_constraints(),
same_unary_allocation=source_problem.get_same_unary_allocation(),
optional_tasks=optional_tasks,
list_objective_computer=list_objective_computer,
alternative_scheduling_subproblems=source_problem.get_alternative_scheduling_subproblem(),
)
[docs]
def forward_transform_solution(
self,
solution: SpecificSchedulingSolution,
target_problem: GenericSchedulingImplProblem,
) -> Optional[GenericSchedulingImplSolution]:
return convert_solution_from_specific_to_generic(
solution=solution,
generic_problem=target_problem,
)
def _construct_min_only_time_lags_from_min_and_max_time_lags(
source_problem: GenericSchedulingProblem,
task1_start_or_end: StartOrEnd,
task2_start_or_end: StartOrEnd,
) -> list[tuple[Task, Task, int]]:
return source_problem.get_original_time_lags(
task1_start_or_end=task1_start_or_end,
task2_start_or_end=task2_start_or_end,
min_or_max=MinOrMax.MIN,
) + [
(task2, task1, -offset)
for task1, task2, offset in source_problem.get_original_time_lags(
task1_start_or_end=task2_start_or_end,
task2_start_or_end=task1_start_or_end,
min_or_max=MinOrMax.MAX,
)
]
def _extract_resource_consumptions(
source_problem: GenericSchedulingProblem,
) -> tuple[
dict[Task, dict[int, dict[CumulativeResource | NonRenewableResource, int]]],
dict[
Task,
dict[
int,
dict[
CumulativeResource | NonRenewableResource,
dict[frozenset[tuple[Task, int]], int],
],
],
],
]:
"""Extract static and dependent resource consumptions from a scheduling problem.
Returns:
Tuple of (static_consumptions, dependent_consumptions)
"""
resources_consumption: dict[
Task, dict[int, dict[CumulativeResource | NonRenewableResource, int]]
] = {}
resources_consumption_dependent: dict[
Task,
dict[
int,
dict[
CumulativeResource | NonRenewableResource,
dict[frozenset[tuple[Task, int]], int],
],
],
] = {}
for task in source_problem.tasks_list:
resources_consumption[task] = {}
resources_consumption_dependent[task] = {}
for mode in source_problem.get_task_modes(task=task):
resources_consumption[task][mode] = {}
resources_consumption_dependent[task][mode] = {}
# Non-renewable resources
for resource in source_problem.non_renewable_resources_list:
if source_problem.is_non_renewable_resource_task_mode_consumption_dependent(
resource=resource, task=task, mode=mode
):
resources_consumption_dependent[task][mode][resource] = (
source_problem.get_non_renewable_resource_consumption_mapping(
resource=resource, task=task, mode=mode
)
)
else:
conso = source_problem.get_non_renewable_resource_consumption(
resource=resource, task=task, mode=mode
)
if conso != 0:
resources_consumption[task][mode][resource] = conso
# Cumulative resources
for resource in source_problem.cumulative_resources_list:
if source_problem.is_cumulative_resource_task_mode_consumption_dependent(
resource=resource, task=task, mode=mode
):
resources_consumption_dependent[task][mode][resource] = (
source_problem.get_cumulative_resource_consumption_mapping(
resource=resource, task=task, mode=mode
)
)
else:
conso = source_problem.get_cumulative_resource_consumption(
resource=resource, task=task, mode=mode
)
if conso != 0:
resources_consumption[task][mode][resource] = conso
return resources_consumption, resources_consumption_dependent
[docs]
def convert_solution_from_specific_to_generic(
solution: SpecificSchedulingSolution,
generic_problem: GenericSchedulingImplProblem,
) -> GenericSchedulingImplSolution:
return GenericSchedulingImplSolution(
problem=generic_problem,
raw_sol=RawSolution(
task_variables={
task: TaskVariable(
start=solution.get_start_time(task),
end=solution.get_end_time(task),
mode=solution.get_mode(task),
is_present=solution.is_present(task),
allocated={
unary_resource: {
skill
for skill in generic_problem.skills_list
if solution.is_skill_used(
task=task,
skill=skill,
unary_resource=unary_resource,
)
}
for unary_resource in solution.get_task_allocation(task)
},
)
for task in generic_problem.tasks_list
}
),
)
[docs]
class FromGenericSchedulingImpl(
ProblemTransformation[
GenericSchedulingImplProblem,
GenericSchedulingImplSolution,
SpecificSchedulingProblem,
SpecificSchedulingSolution,
],
Generic[SpecificSchedulingProblem, SpecificSchedulingSolution],
):
"""Transform the generic implementation of a scheduling problem into a specific one.
This is still an abstract class, as the `transform_problem()` remains to implement.
"""
[docs]
def back_transform_solution(
self,
solution: SpecificSchedulingSolution,
source_problem: GenericSchedulingImplProblem,
) -> GenericSchedulingImplSolution:
return convert_solution_from_specific_to_generic(
solution=solution,
generic_problem=source_problem,
)