# 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 __future__ import annotations
from collections.abc import Container, Hashable, Iterable
from copy import deepcopy
from dataclasses import InitVar, dataclass, field
from typing import Callable, Optional
import numpy as np
import wrapt
from discrete_optimization.generic_tasks_tools.alternative_subproblems import (
AlternativeSchedulingSubProblem,
)
from discrete_optimization.generic_tasks_tools.calendar_resource import (
convert_availability_intervals_to_calendar,
convert_calendar_to_availability_intervals,
)
from discrete_optimization.generic_tasks_tools.enums import (
AbsentValue,
MinOrMax,
StartOrEnd,
)
from discrete_optimization.generic_tasks_tools.generic_scheduling import (
GenericSchedulingProblem,
GenericSchedulingSolution,
)
from discrete_optimization.generic_tasks_tools.generic_scheduling_utils import (
OBJECTIVE_DEFAULT_WEIGHTS,
Objective,
RawSolution,
)
from discrete_optimization.generic_tasks_tools.multimode import ModeConstraintType
from discrete_optimization.generic_tasks_tools.objectives.makespan import (
MakespanObjectiveComputer,
)
from discrete_optimization.generic_tasks_tools.objectives.objective_computer import (
ObjectiveComputer,
)
from discrete_optimization.generic_tasks_tools.resource_blocking import (
FlexibleGapBlockingConstraint,
SpanBlockingConstraint,
)
from discrete_optimization.generic_tools.do_problem import (
Solution,
)
from discrete_optimization.generic_tools.encoding_register import EncodingRegister
# types for annotations
Skill = Hashable
NonSkillCumulativeResource = Hashable
NonRenewableResource = Hashable
UnaryResource = Hashable
Task = Hashable
CumulativeResource = NonSkillCumulativeResource | Skill
Resource = CumulativeResource | UnaryResource # calendar resources
AnyResource = NonRenewableResource | Resource
UnaryAvailabilityIntervals = list[tuple[int, int]] # start, end
AvailabilityIntervals = list[tuple[int, int, int]] # start, end, value
[docs]
@dataclass
class GenericSchedulingImplProblem(
GenericSchedulingProblem[
Task, UnaryResource, Skill, NonSkillCumulativeResource, NonRenewableResource
]
):
"""Generic implementation of a scheduling problem.
It implements the abstract class `GenericSchedulingProblem`.
Attributes:
horizon: max allowed time to finish the tasks
durations_per_mode: task -> mode -> duration. Tasks durations, mode by mode.
This is used to know all available tasks, all available modes for a given task, and corresponding durations.
resource_consumptions: task -> mode -> resource -> conso.
Cumulative or non-renewable resource consumption, task by task, mode by mode. The resource can be a skill.
Missing key => conso = 0
successors: maps a task to its successors in the precedence graph.
Each successor task must start after the given task ends.
Default to no precedence constraints. Note that a consolidated version of it will
be constructed using the time lags constraints.
unary_resources: available unary resources. Default to none.
unary_resources_skills: skill values of each unary resource. Missing key => skill value = 0
unary_resources_availabilities: availability of unary resources on the form of list of intervals (start, end)
Missing key => always available.
unary_resources_task_compatibility: maps a task to its compatible unary resources. Missing key => all unary resources allowed.
skills: available skills
non_skill_cumulative_resources: cumulative resources (excluding skills) availabilities.
Format: either int => always available at the given max capacity,
or list of intervals + capacity (start, end, value)
non_renewable_resources: non-renewable resources max capacities
time_windows: maps task to start_lb, end_lb, start_ub, end_ub s.t.
start_lb <= start(task) <= start_ub and end_lb <= end(task) <= end_ub
missing or none value means 0 (lb) or self.horizon (ub)
start_to_start_min_time_lags: min time lags constraints between task starts.
task1, task2, offset meaning start(task1) + offset <= start(task2)
Note that using negative offset can model start-to-start max time lags.
start_to_end_min_time_lags: min time lags constraints first task start and second task end.
task1, task2, offset meaning start(task1) + offset <= end(task2)
Note that using negative offset can model end-to-start max time lags.
end_to_start_min_time_lags: min time lags constraints between first task end and second task start.
task1, task2, offset meaning end(task1) + offset <= start(task2)
Note that using negative offset can model start-to-end max time lags.
end_to_end_min_time_lags: min time lags constraints between task ends.
task1, task2, offset meaning end(task1) + offset <= end(task2)
Note that using negative offset can model end-to-end max time lags.
no_overlap_sets: a set of (set of tasks that should not overlap together)
forbidden_intervals: maps task to forbidden intervals that cannot overlap with it. Missing key => no forbidden intervals.
flexible_gap_blocking_constraints: list of blocking constraints between entity points.
Each constraint is (entity1, point1, entity2, point2, resources, metadata).
Default to no blocking constraints.
span_blocking_constraints: list of span blocking constraints.
Each constraint is (tasks, resources, metadata) where tasks is a frozenset.
Default to no blocking constraints.
objective: objective for the problem. Default to minimization of makespan.
Either an iterable of (objective, weight) so that the problem should *maximize* the aggregated objective
resulting from weighted sum of objectives, or a single objective in which case we use the corresponding
default weight from
`discrete_optimization.generic_tasks_tools.generic_scheduling_utils.OBJECTIVE_DEFAULT_WEIGHTS`
and maximize it. For instance the default weight for makespan is -1 so that it will
actually minimize the makespan.
custom_evaluate_fn: function used to evaluate the "custom" objective (to be maximized).
objective_resource_weights: Weights to be used by the objective when summing used resources
(`Objective.NB_RESOURCES_USED`) or resources levels (`Objective.RESOURCES_LEVELS`).
Default to 1 for resources not mentioned.
mode_costs: cost of choosing each mode. Missing key => cost = 0.
unary_resource_costs: cost of allocating each unary resource. Missing key => cost = 0.
compute_time_penalty: whether to include time penalties in evaluation
"""
horizon: int
durations_per_mode: dict[Task, dict[int, int]]
resource_consumptions: dict[
Task, dict[int, dict[CumulativeResource | NonRenewableResource, int]]
] = field(default_factory=dict)
resource_consumptions_dependent: dict[
Task,
dict[
int,
dict[
CumulativeResource | NonRenewableResource,
dict[frozenset[tuple[Task, int]], int],
],
],
] = field(default_factory=dict)
successors: dict[Task, set[Task]] = field(default_factory=dict)
unary_resources: set[UnaryResource] = field(default_factory=set)
unary_resources_skills: dict[UnaryResource, dict[Skill, int]] = field(
default_factory=dict
)
unary_resources_availabilities: dict[UnaryResource, UnaryAvailabilityIntervals] = (
field(default_factory=dict)
)
unary_resources_task_compatibility: dict[Task, set[UnaryResource]] = field(
default_factory=dict
)
skills: set[Skill] = field(default_factory=set)
non_skill_cumulative_resources: dict[
NonSkillCumulativeResource, int | AvailabilityIntervals
] = field(default_factory=dict)
non_renewable_resources: dict[NonRenewableResource, int] = field(
default_factory=dict
)
time_windows: dict[Task, tuple[int | None, int | None, int | None, int | None]] = (
field(default_factory=dict)
)
start_to_start_min_time_lags: list[tuple[Task, Task, int]] = field(
default_factory=list
)
start_to_end_min_time_lags: list[tuple[Task, Task, int]] = field(
default_factory=list
)
end_to_start_min_time_lags: list[tuple[Task, Task, int]] = field(
default_factory=list
)
end_to_end_min_time_lags: list[tuple[Task, Task, int]] = field(default_factory=list)
no_overlap_sets: set[frozenset[Task]] = field(default_factory=set)
forbidden_intervals: dict[Task, list[tuple[int, int]]] = field(default_factory=dict)
flexible_gap_blocking_constraints: list[FlexibleGapBlockingConstraint] = field(
default_factory=list
)
span_blocking_constraints: list[SpanBlockingConstraint] = field(
default_factory=list
)
mode_constraints: list[tuple[ModeConstraintType, list[tuple[Task, int]]]] = field(
default_factory=list
)
same_unary_allocation: list[set[Task]] = field(default_factory=list)
objective: InitVar[Objective | Iterable[tuple[Objective, int]]] = Objective.MAKESPAN
custom_evaluate_fn: Optional[Callable[[GenericSchedulingImplSolution], int]] = None
objective_resource_weights: Optional[dict[AnyResource, int]] = None
mode_costs: dict[Task, dict[int, int]] = field(default_factory=dict)
unary_resource_costs: dict[Task, dict[int, dict[UnaryResource, int]]] = field(
default_factory=dict
)
compute_time_penalty: bool = True
optional_tasks: set[Task] = field(default_factory=set)
list_objective_computer: list[ObjectiveComputer] = field(
compare=False, default=None
)
alternative_scheduling_subproblems: list[AlternativeSchedulingSubProblem] = field(
default_factory=list
)
def __post_init__(self, objective: Objective | Iterable[tuple[Objective, int]]):
if self.list_objective_computer is None:
self.list_objective_computer = [
MakespanObjectiveComputer(
None, OBJECTIVE_DEFAULT_WEIGHTS[Objective.MAKESPAN]
)
]
for l in self.list_objective_computer:
l.set_problem(self)
self.update_problem()
[docs]
def evaluate(self, variable: GenericSchedulingSolution) -> dict[str, float]:
kpis = super().evaluate(variable)
if self.custom_evaluate_fn is not None:
kpis[Objective.CUSTOM] = self.custom_evaluate_fn(variable)
return kpis
[docs]
def update_problem(self):
"""Method to call when some attributes of the problem are modified."""
self._tasks_list = list(self.durations_per_mode)
self._skills_list = list(self.skills)
self._non_skill_cumulative_resources_list = list(
self.non_skill_cumulative_resources
)
self._non_renewable_resources_list = list(self.non_renewable_resources)
self._unary_resources_list = list(self.unary_resources)
self.check_resources_lists()
self.update_tasks_list()
self.update_skills()
self.update_resource_availabilities()
self.update_task_bounds()
self.update_time_lags()
self.update_precedence_constraints()
[docs]
def update_resource_availabilities(self) -> None:
self.get_resource_availabilities.cache_clear()
super().update_resource_availabilities()
[docs]
def check_resources_lists(self) -> None:
"""Check duplicates in resources."""
resources_list = (
self.calendar_resources_list + self.non_renewable_resources_list
)
assert len(resources_list) == len(set(resources_list)), (
"There are duplicates in resources list, "
"potentially because calendar and non-renewable resources intersect."
)
[docs]
def get_alternative_scheduling_subproblem(
self,
) -> list[AlternativeSchedulingSubProblem]:
return self.alternative_scheduling_subproblems
[docs]
def is_cumulative_resource_task_mode_consumption_dependent(
self, resource: CumulativeResource, task: Task, mode: int
) -> bool:
try:
return resource in self.resource_consumptions_dependent[task][mode]
except KeyError:
return False
[docs]
def get_cumulative_resource_consumption_mapping(
self, resource: CumulativeResource, task: Task, mode: int
) -> dict[frozenset[tuple[Task, int]], int]:
if self.is_cumulative_resource_task_mode_consumption_dependent(
resource, task, mode
):
return self.resource_consumptions_dependent[task][mode][resource]
return super().get_cumulative_resource_consumption_mapping(resource, task, mode)
[docs]
def is_non_renewable_resource_task_mode_consumption_dependent(
self, resource: NonRenewableResource, task: Task, mode: int
) -> bool:
try:
return resource in self.resource_consumptions_dependent[task][mode]
except KeyError:
return False
[docs]
def get_non_renewable_resource_consumption_mapping(
self, resource: CumulativeResource, task: Task, mode: int
) -> dict[frozenset[tuple[Task, int]], int]:
if self.is_non_renewable_resource_task_mode_consumption_dependent(
resource, task, mode
):
return self.resource_consumptions_dependent[task][mode][resource]
return super().get_non_renewable_resource_consumption_mapping(
resource, task, mode
)
@property
def skills_list(self) -> list[Skill]:
return self._skills_list
@property
def non_skill_cumulative_resources_list(self) -> list[Skill]:
return self._non_skill_cumulative_resources_list
[docs]
def get_unary_resource_skill_value(
self, unary_resource: UnaryResource, skill: Skill
) -> int:
try:
return self.unary_resources_skills[unary_resource][skill]
except KeyError:
return 0
[docs]
def is_compatible_task_unary_resource(
self, task: Task, unary_resource: UnaryResource
) -> bool:
compatible_unary_resources = self.unary_resources_task_compatibility.get(
task, None
)
if compatible_unary_resources is None:
return super().is_compatible_task_unary_resource(task, unary_resource)
else:
return unary_resource in compatible_unary_resources
[docs]
def get_cumulative_resource_consumption(
self, resource: CumulativeResource, task: Task, mode: int
) -> int:
try:
return self.resource_consumptions[task][mode][resource]
except KeyError:
return 0
[docs]
def get_no_overlap(self) -> set[frozenset[Task]]:
return self.no_overlap_sets
[docs]
def get_forbidden_intervals(self, task: Task) -> list[tuple[int, int]]:
return self.forbidden_intervals.get(task, [])
[docs]
def get_flexible_gap_blocking_constraints(
self,
) -> list[FlexibleGapBlockingConstraint]:
"""Return flexible gap blocking constraints."""
return self.flexible_gap_blocking_constraints
[docs]
def get_span_blocking_constraints(self) -> list[SpanBlockingConstraint]:
"""Return span blocking constraints."""
return self.span_blocking_constraints
[docs]
def get_mode_constraints(
self,
) -> list[tuple[ModeConstraintType, list[tuple[Task, int]]]]:
return self.mode_constraints
[docs]
def get_same_unary_allocation(self) -> list[set[Task]]:
return self.same_unary_allocation
[docs]
@wrapt.lru_cache(maxsize=None)
def get_resource_availabilities(
self, resource: Resource
) -> list[tuple[int, int, int]]:
if resource in self.skills:
return self.compute_skill_availabilities(resource)
elif resource in self.unary_resources:
try:
return [
(start, end, 1)
for start, end in self.unary_resources_availabilities[resource]
]
except KeyError:
return [(0, self.horizon, 1)]
elif resource in self.non_skill_cumulative_resources:
intervals_or_max_capacity = self.non_skill_cumulative_resources[resource]
if isinstance(intervals_or_max_capacity, int):
return [(0, self.horizon, intervals_or_max_capacity)]
else:
return intervals_or_max_capacity
else:
raise ValueError(
f"{resource} is neither an actual cumulative resource, nor a skill, nor a unary resource."
)
[docs]
def get_task_mode_duration(self, task: Task, mode: int) -> int:
return self.durations_per_mode[task][mode]
@property
def non_renewable_resources_list(self) -> list[NonRenewableResource]:
return self._non_renewable_resources_list
[docs]
def get_non_renewable_resource_capacity(
self, resource: NonRenewableResource
) -> int:
return self.non_renewable_resources[resource]
[docs]
def get_non_renewable_resource_consumption(
self, resource: NonRenewableResource, task: Task, mode: int
) -> int:
try:
return self.resource_consumptions[task][mode][resource]
except KeyError:
return 0
[docs]
def get_start_to_start_min_time_lags(self) -> list[tuple[Task, Task, int]]:
return self.start_to_start_min_time_lags
[docs]
def get_end_to_start_min_time_lags(self) -> list[tuple[Task, Task, int]]:
return self.end_to_start_min_time_lags
[docs]
def get_end_to_end_min_time_lags(self) -> list[tuple[Task, Task, int]]:
return self.end_to_end_min_time_lags
[docs]
def get_start_to_end_min_time_lags(self) -> list[tuple[Task, Task, int]]:
return self.start_to_end_min_time_lags
[docs]
def get_task_start_or_end_lower_bound(
self, task: Task, start_or_end: StartOrEnd
) -> int:
try:
start_lb, end_lb, start_ub, end_ub = self.time_windows[task]
if start_or_end == StartOrEnd.START:
lb = start_lb
else:
lb = end_lb
if lb is not None:
return lb
except KeyError:
pass
return super().get_task_start_or_end_lower_bound(task, start_or_end)
[docs]
def get_task_start_or_end_upper_bound(
self, task: Task, start_or_end: StartOrEnd
) -> int:
try:
start_lb, end_lb, start_ub, end_ub = self.time_windows[task]
if start_or_end == StartOrEnd.START:
ub = start_ub
else:
ub = end_ub
if ub is not None:
return ub
except KeyError:
pass
return super().get_task_start_or_end_upper_bound(task, start_or_end)
[docs]
def get_precedence_constraints(self) -> dict[Task, Iterable[Task]]:
return self.successors
[docs]
def get_makespan_upper_bound(self) -> int:
return self.horizon
[docs]
def get_task_modes(self, task: Task) -> set[int]:
return set(self.durations_per_mode[task])
@property
def unary_resources_list(self) -> list[UnaryResource]:
return self._unary_resources_list
@property
def tasks_list(self) -> list[Task]:
return self._tasks_list
[docs]
def is_optional(self, task: Task) -> bool:
return task in self.optional_tasks
[docs]
def get_solution_type(self) -> type[Solution]:
return GenericSchedulingImplSolution
[docs]
def get_attribute_register(self) -> EncodingRegister:
raise NotImplementedError()
[docs]
def set_fixed_attributes(self, attribute_name: str, solution: Solution) -> None:
raise NotImplementedError()
[docs]
def get_list_objective_computer(self) -> list[ObjectiveComputer]:
return self.list_objective_computer
[docs]
def get_dummy_solution(self) -> Solution:
raise NotImplementedError()
[docs]
def create_subproblem_from_partial_solution(
self, partial_solution: RawSolution[Task, UnaryResource, Skill]
) -> GenericSchedulingImplProblem:
"""Create a subproblem according to a partial solution.
- Tasks already scheduled are removed from subproblem.
- Add time windows constraints to model timelags/precedence constraints with removed tasks.
- Update resource calendars with scheduled allocations
- Update non-renewable resources max capacities
- Transform no_overlap constraints into forbidden intervals constraints
"""
scheduled_tasks = {
task: task_variable
for task, task_variable in partial_solution.task_variables.items()
if task_variable.is_present
}
# restrict tasks list
new_tasks_list = [
task for task in self.tasks_list if task not in scheduled_tasks
]
new_durations_per_mode = {
task: self.durations_per_mode[task] for task in new_tasks_list
}
new_successors = {
task: {
next_task
for next_task in next_tasks
if next_task not in scheduled_tasks
}
for task, next_tasks in self.successors.items()
if task not in scheduled_tasks
}
new_start_to_start_min_time_lags = _restrict_timelags(
self.start_to_start_min_time_lags, tasks_to_remove=scheduled_tasks
)
new_end_to_start_min_time_lags = _restrict_timelags(
self.end_to_start_min_time_lags, tasks_to_remove=scheduled_tasks
)
new_start_to_end_min_time_lags = _restrict_timelags(
self.start_to_end_min_time_lags, tasks_to_remove=scheduled_tasks
)
new_end_to_end_min_time_lags = _restrict_timelags(
self.end_to_end_min_time_lags, tasks_to_remove=scheduled_tasks
)
# translate time lags/precedence constraints involving missing tasks into time windows constraints
new_time_windows_nested: dict[tuple[Task, StartOrEnd, MinOrMax], set[int]] = {
(task, start_or_end, min_or_max): {
# at least the original bound
self.get_task_bound(
task=task,
start_or_end=start_or_end,
min_or_max=min_or_max,
)
}
for task in new_tasks_list
for start_or_end in StartOrEnd
for min_or_max in MinOrMax
}
for task1_start_or_end in StartOrEnd:
for task2_start_or_end in StartOrEnd:
for min_or_max in MinOrMax:
for task1, task2, offset in self.get_consolidated_time_lags(
task1_start_or_end=task1_start_or_end,
task2_start_or_end=task2_start_or_end,
min_or_max=min_or_max,
):
if task1 in scheduled_tasks and task2 not in scheduled_tasks:
scheduled_task = task1
task = task2
scheduled_task_start_or_end = task1_start_or_end
task_start_or_end = task2_start_or_end
scheduled_offset = offset
task_min_or_max = min_or_max
elif task1 not in scheduled_tasks and task2 in scheduled_tasks:
scheduled_task = task2
task = task1
scheduled_task_start_or_end = task2_start_or_end
task_start_or_end = task1_start_or_end
scheduled_offset = -offset
task_min_or_max = ~min_or_max
else:
continue
bound_from_schedule = (
scheduled_tasks[scheduled_task].get_start_or_end(
scheduled_task_start_or_end
)
+ scheduled_offset
)
new_time_windows_nested[
(task, task_start_or_end, task_min_or_max)
].add(bound_from_schedule)
new_time_windows = {
task: (
max(new_time_windows_nested[(task, StartOrEnd.START, MinOrMax.MIN)]),
max(new_time_windows_nested[(task, StartOrEnd.END, MinOrMax.MIN)]),
min(new_time_windows_nested[(task, StartOrEnd.START, MinOrMax.MAX)]),
min(new_time_windows_nested[(task, StartOrEnd.END, MinOrMax.MAX)]),
)
for task in new_tasks_list
}
# Update non-renewable resources max capacities
new_non_renewable_resources = {
resource: old_max_capacity
- sum(
self.get_non_renewable_resource_consumption(
resource=resource, task=task, mode=task_variable.mode
)
for task, task_variable in scheduled_tasks.items()
)
for resource, old_max_capacity in self.non_renewable_resources.items()
}
# Update resources availabilities
calendar_resources = (
self.unary_resources_list + self.non_skill_cumulative_resources_list
)
resources_calendars = {
resource: np.array(
convert_availability_intervals_to_calendar(
intervals=self.get_resource_availabilities(resource=resource),
horizon=self.horizon,
),
dtype=int,
)
for resource in calendar_resources
}
for task, task_variable in scheduled_tasks.items():
start = task_variable.start
end = task_variable.end
for resource in self.unary_resources_list:
if resource in task_variable.allocated:
resources_calendars[resource][start:end] -= 1
for resource in self.non_skill_cumulative_resources_list:
conso = self.get_cumulative_resource_consumption(
resource=resource, task=task, mode=task_variable.mode
)
resources_calendars[resource][start:end] -= conso
new_non_skill_cumulative_resources = {
resource: convert_calendar_to_availability_intervals(
calendar=resources_calendars[resource],
horizon=self.horizon,
)
for resource in self.non_skill_cumulative_resources_list
}
new_unary_resources_availabilities = {
resource: [
(start, end)
for start, end, value in convert_calendar_to_availability_intervals(
calendar=resources_calendars[resource],
horizon=self.horizon,
)
if value > 0
]
for resource in self.unary_resources_list
}
# Transform no_overlap constraints into forbidden intervals constraints
new_no_overlap_sets: set[frozenset[Task]] = set()
new_forbidden_intervals: dict[Task, list[tuple[int, int]]] = {
task: list(self.get_forbidden_intervals(task=task))
for task in self.tasks_list
}
for not_overlapping_tasks in self.get_no_overlap():
if not_overlapping_tasks.isdisjoint(scheduled_tasks):
# no removed tasks in it => ok
new_no_overlap_sets.add(not_overlapping_tasks)
else:
# remove scheduled tasks and replace them by forbidden intervals
new_not_overlapping_tasks = not_overlapping_tasks.difference(
scheduled_tasks
)
scheduled_tasks_in_not_overlapping_tasks = (
not_overlapping_tasks.difference(new_not_overlapping_tasks)
)
new_no_overlap_sets.add(new_not_overlapping_tasks)
for task in new_not_overlapping_tasks:
new_forbidden_intervals[task].extend(
[
(
scheduled_tasks[scheduled_task].start,
scheduled_tasks[scheduled_task].end,
)
for scheduled_task in scheduled_tasks_in_not_overlapping_tasks
]
)
return GenericSchedulingImplProblem(
horizon=self.horizon,
durations_per_mode=new_durations_per_mode,
resource_consumptions=self.resource_consumptions,
resource_consumptions_dependent=self.resource_consumptions_dependent,
successors=new_successors,
unary_resources=self.unary_resources,
unary_resources_skills=self.unary_resources_skills,
unary_resources_availabilities=new_unary_resources_availabilities,
unary_resources_task_compatibility=self.unary_resources_task_compatibility,
skills=self.skills,
non_skill_cumulative_resources=new_non_skill_cumulative_resources,
non_renewable_resources=new_non_renewable_resources,
time_windows=new_time_windows,
start_to_start_min_time_lags=new_start_to_start_min_time_lags,
start_to_end_min_time_lags=new_start_to_end_min_time_lags,
end_to_start_min_time_lags=new_end_to_start_min_time_lags,
end_to_end_min_time_lags=new_end_to_end_min_time_lags,
no_overlap_sets=new_no_overlap_sets,
forbidden_intervals=new_forbidden_intervals,
list_objective_computer=self.list_objective_computer,
)
def _restrict_timelags(
timelags: list[tuple[Task, Task, int]], tasks_to_remove: Container[Task]
) -> list[tuple[Task, Task, int]]:
return [
(task1, task2, offset)
for task1, task2, offset in timelags
if (task1 not in tasks_to_remove) and (task2 not in tasks_to_remove)
]
[docs]
class GenericSchedulingImplSolution(
GenericSchedulingSolution[
Task, UnaryResource, Skill, NonSkillCumulativeResource, NonRenewableResource
]
):
"""Generic implementation of a solution to a scheduling problem.
It implements the abstract class `GenericSchedulingSolution`.
"""
problem: GenericSchedulingImplProblem
def __init__(
self,
problem: GenericSchedulingImplProblem,
raw_sol: RawSolution[Task, UnaryResource, Skill],
):
super().__init__(problem)
self.raw_sol = raw_sol
[docs]
def is_skill_used(
self, task: Task, unary_resource: UnaryResource, skill: Skill
) -> bool:
try:
return skill in self.raw_sol.task_variables[task].allocated[unary_resource]
except KeyError:
return False
[docs]
def get_end_time(self, task: Task) -> int | AbsentValue:
if task not in self.raw_sol.task_variables:
return AbsentValue.ABSENT
return self.raw_sol.task_variables[task].end
[docs]
def get_start_time(self, task: Task) -> int | AbsentValue:
if task not in self.raw_sol.task_variables:
return AbsentValue.ABSENT
return self.raw_sol.task_variables[task].start
[docs]
def get_mode(self, task: Task) -> int | AbsentValue:
if task not in self.raw_sol.task_variables:
return AbsentValue.ABSENT
return self.raw_sol.task_variables[task].mode
[docs]
def is_present(self, task: Task) -> bool:
if task not in self.raw_sol.task_variables:
return False
return self.raw_sol.task_variables[task].is_present
[docs]
def is_allocated(self, task: Task, unary_resource: UnaryResource) -> bool:
if task not in self.raw_sol.task_variables:
return False
return unary_resource in self.raw_sol.task_variables[task].allocated
[docs]
def get_task_allocation(self, task: Task) -> set[UnaryResource]:
if task not in self.raw_sol.task_variables:
return set()
return set(self.raw_sol.task_variables[task].allocated)
[docs]
def copy(self) -> Solution:
return GenericSchedulingImplSolution(
problem=self.problem, raw_sol=deepcopy(self.raw_sol)
)
[docs]
def lazy_copy(self) -> Solution:
return GenericSchedulingImplSolution(problem=self.problem, raw_sol=self.raw_sol)