Source code for discrete_optimization.rcpsp_multiskill.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.
"""Transformation from/to generic scheduling to/from RCPSP Multiskill."""
from __future__ import annotations
import itertools
from collections.abc import Hashable
from discrete_optimization.generic_tasks_tools.enums import StartOrEnd
from discrete_optimization.generic_tasks_tools.generic_scheduling_impl import (
GenericSchedulingImplProblem,
Skill,
Task,
UnaryResource,
)
from discrete_optimization.generic_tasks_tools.generic_scheduling_utils import (
RawSolution,
)
from discrete_optimization.generic_tasks_tools.transformations.generic_scheduling_impl import (
FromGenericSchedulingImpl,
ToGenericSchedulingImpl,
)
from discrete_optimization.generic_tools.transformation import (
InformationLoss,
LossImpact,
LossType,
TransformationMetadata,
lossy_transformation,
)
from discrete_optimization.rcpsp.special_constraints import (
SpecialConstraintsDescription,
)
from discrete_optimization.rcpsp_multiskill.problem import (
NB_EMPLOYEES_LB,
Employee,
MultiskillRcpspProblem,
MultiskillRcpspSolution,
SkillDetail,
)
[docs]
def transform_solution_from_raw_generic_to_rcpsp_ms(
raw_sol: RawSolution[Task, UnaryResource, Skill], problem: MultiskillRcpspProblem
) -> MultiskillRcpspSolution:
"""Convert generic solution to RCPSP multiskill solution.
Args:
solution:
problem:
Returns:
"""
modes_dict = {}
schedule = {}
employee_usage = {}
for task, task_variable in raw_sol.task_variables.items():
schedule[task] = {
"start_time": task_variable.start,
"end_time": task_variable.end,
}
modes_dict[task] = task_variable.mode
employee_usage[task] = {
worker: skills
for worker, skills in task_variable.allocated.items()
if len(skills) > 0
}
return MultiskillRcpspSolution(
problem=problem,
schedule=schedule,
modes=modes_dict,
employee_usage=employee_usage,
)
[docs]
class MultiskillRcpspToGenericSchedulingTransformation(
ToGenericSchedulingImpl[MultiskillRcpspProblem, MultiskillRcpspSolution]
):
"""Transform RCPSP multiskill to GenericSchedulingImplProblem."""
[docs]
def transform_solution_from_raw_generic_to_specific(
self,
raw_sol: RawSolution[Task, UnaryResource, Skill],
source_problem: MultiskillRcpspProblem,
) -> MultiskillRcpspSolution:
return transform_solution_from_raw_generic_to_rcpsp_ms(
raw_sol=raw_sol, problem=source_problem
)
[docs]
def get_forward_metadata(self) -> TransformationMetadata:
return lossy_transformation(
losses=[
InformationLoss(
name="task_mode",
loss_type=LossType.CONSTRAINT,
description="fixed modes by tasks in special constraints.",
impact=LossImpact.MAJOR,
reason="not possible with generic scheduling implementation.",
),
InformationLoss(
name="pair_mode_constraint",
loss_type=LossType.CONSTRAINT,
description="constraints on pair of tasks that must share a mode or a score.",
impact=LossImpact.MAJOR,
reason="not possible with generic scheduling implementation.",
),
]
)
[docs]
class GenericSchedulingToMultiskillRcpspTransformation(
FromGenericSchedulingImpl[MultiskillRcpspProblem, MultiskillRcpspSolution]
):
"""Transform GenericSchedulingImplProblem to RCPSP multiskill."""
[docs]
def get_forward_metadata(self) -> TransformationMetadata:
return lossy_transformation(
losses=[
InformationLoss(
name="forbidden_intervals",
loss_type=LossType.CONSTRAINT,
description="constraints on fixed intervals with which a given task cannot overlap.",
impact=LossImpact.MAJOR,
reason="rcpsp special constraints cannot handle it.",
),
InformationLoss(
name="time_lags",
loss_type=LossType.CONSTRAINT,
description="time lags constraints, but only end_to_start max, start_to_end min, end_to_end min/max.",
impact=LossImpact.MAJOR,
reason="rcpsp special constraints can only handle the other time lags constraints.",
),
InformationLoss(
name="objective",
loss_type=LossType.OBJECTIVE,
description="whenever the objective is not just makespan",
impact=LossImpact.MODERATE,
reason="rcpsp has fixed objective.",
),
]
)
[docs]
def transform_problem(
self, source_problem: GenericSchedulingImplProblem
) -> MultiskillRcpspProblem:
"""
Args:
source_problem:
Returns:
"""
horizon: int = source_problem.horizon
resources_availability: dict[str, list[int]] = {
str(resource): [
source_problem.get_non_renewable_resource_capacity(resource=resource)
]
for resource in source_problem.non_renewable_resources_list
} | {
str(resource): source_problem.get_resource_calendar(resource=resource)
for resource in source_problem.non_skill_cumulative_resources_list
if resource != NB_EMPLOYEES_LB
}
non_renewable_resources: set[str] = {
str(resource) for resource in source_problem.non_renewable_resources_list
}
resources_set: set[str] = set(resources_availability)
skills_set: set[str] = {str(skill) for skill in source_problem.skills_list}
employees: dict[UnaryResource, Employee] = {
unary_resource: Employee(
dict_skill={
str(skill): SkillDetail(
skill_value=value, efficiency_ratio=1, experience=0
)
for skill in source_problem.skills_list
if (
(
value := source_problem.get_unary_resource_skill_value(
unary_resource=unary_resource, skill=skill
)
)
> 0
)
},
calendar_employee=[
bool(value)
for value in source_problem.get_resource_calendar(
resource=unary_resource
)
],
)
for unary_resource in source_problem.unary_resources_list
}
mode_details: dict[Hashable, dict[int, dict[str, int]]] = {
task: {
mode: {
str(
resource
): source_problem.get_non_renewable_resource_consumption(
resource=resource, task=task, mode=mode
)
for resource in source_problem.non_renewable_resources_list
}
| {
str(resource): source_problem.get_cumulative_resource_consumption(
resource=resource, task=task, mode=mode
)
for resource in source_problem.cumulative_resources_list
if resource != NB_EMPLOYEES_LB
}
| {
"duration": source_problem.get_task_mode_duration(
task=task, mode=mode
)
}
for mode in source_problem.get_task_modes(task)
}
for task in source_problem.tasks_list
}
successors = {
task: list(next_tasks)
for task, next_tasks in source_problem.get_precedence_constraints().items()
}
# find source task and sink task or create them
precedence_graph = source_problem.get_precedence_graph()
first_tasks = [
task
for task, n_predecessors in precedence_graph.graph_nx.in_degree
if n_predecessors == 0
]
if len(first_tasks) > 1:
# add a source task
source_task = "source"
if source_task in mode_details or source_task in successors:
raise ValueError(
f"The source problem does not have a source task but has a task named '{source_task}'."
)
mode_details[source_task] = {1: {"duration": 0}}
successors[source_task] = first_tasks
else:
source_task = first_tasks[0]
last_tasks = [
task
for task, n_successors in precedence_graph.graph_nx.out_degree
if n_successors == 0
]
if len(last_tasks) > 1:
# add a sink task
sink_task = "sink"
if sink_task in mode_details or sink_task in successors:
raise ValueError(
f"The source problem does not have a sink task but has a task named '{sink_task}'."
)
mode_details[sink_task] = {1: {"duration": 0}}
for task in last_tasks:
successors[task] = [sink_task]
else:
sink_task = last_tasks[0]
# special constraints
# time windows
start_times_window = {
task: (
source_problem.get_task_start_or_end_lower_bound(
task=task, start_or_end=StartOrEnd.START
),
source_problem.get_task_start_or_end_upper_bound(
task=task, start_or_end=StartOrEnd.START
),
)
for task in source_problem.tasks_list
}
end_times_window = {
task: (
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.END
),
)
for task in source_problem.tasks_list
}
# time lags
start_after_end_plus_offset = (
source_problem.get_end_to_start_min_time_lags()
+ [
(t2, t1, -offset)
for t1, t2, offset in source_problem.get_start_to_end_max_time_lags()
]
)
start_to_start_min_time_lag = source_problem.get_start_to_start_min_time_lags()
start_to_start_max_time_lag = source_problem.get_start_to_start_max_time_lags()
# loss: end_to_start max, start_to_end min, end_to_end min/max
# disjunctive tasks
disjunctive_tasks: list[tuple[Task, Task]] = []
for tasks in source_problem.get_no_overlap():
if len(tasks) >= 2:
disjunctive_tasks.extend(itertools.combinations(tasks, 2))
special_constraints = SpecialConstraintsDescription(
start_times_window=start_times_window,
end_times_window=end_times_window,
start_after_end_plus_offset=start_after_end_plus_offset,
start_to_start_min_time_lag=start_to_start_min_time_lag,
start_to_start_max_time_lag=start_to_start_max_time_lag,
disjunctive_tasks=disjunctive_tasks,
)
return MultiskillRcpspProblem(
skills_set=skills_set,
resources_set=resources_set,
non_renewable_resources=non_renewable_resources,
resources_availability=resources_availability,
employees=employees,
mode_details=mode_details,
successors=successors,
horizon=horizon,
source_task=source_task,
sink_task=sink_task,
special_constraints=special_constraints,
only_one_skill_per_task=source_problem.only_one_skill_per_task,
)