discrete_optimization.rcpsp package

Subpackages

Submodules

discrete_optimization.rcpsp.fast_function module

discrete_optimization.rcpsp.fast_function.compute_mean_ressource(modes_array: ndarray[tuple[Any, ...], dtype[int64]], consumption_array: ndarray[tuple[Any, ...], dtype[int64]], start_array: ndarray[tuple[Any, ...], dtype[int64]], end_array: ndarray[tuple[Any, ...], dtype[int64]], horizon: int, ressource_available: ndarray[tuple[Any, ...], dtype[int64]], ressource_renewable: ndarray[tuple[Any, ...], dtype[bool]]) → float[source]
discrete_optimization.rcpsp.fast_function.compute_ressource_consumption(modes_array: ndarray[tuple[Any, ...], dtype[int64]], consumption_array: ndarray[tuple[Any, ...], dtype[int64]], start_array: ndarray[tuple[Any, ...], dtype[int64]], end_array: ndarray[tuple[Any, ...], dtype[int64]], horizon: int, ressource_available: ndarray[tuple[Any, ...], dtype[int64]], ressource_renewable: ndarray[tuple[Any, ...], dtype[bool]]) → dict[int, ndarray[tuple[Any, ...], dtype[int64]]][source]
discrete_optimization.rcpsp.fast_function.sgs_fast(permutation_task: ndarray[tuple[Any, ...], dtype[int64]], modes_array: ndarray[tuple[Any, ...], dtype[int64]], consumption_array: ndarray[tuple[Any, ...], dtype[int64]], duration_array: ndarray[tuple[Any, ...], dtype[int64]], predecessors: ndarray[tuple[Any, ...], dtype[int64]], successors: ndarray[tuple[Any, ...], dtype[int64]], horizon: int, ressource_available: ndarray[tuple[Any, ...], dtype[int64]], ressource_renewable: ndarray[tuple[Any, ...], dtype[bool]], minimum_starting_time_array: ndarray[tuple[Any, ...], dtype[int64]]) → tuple[dict[int, tuple[int, int]], bool][source]
discrete_optimization.rcpsp.fast_function.sgs_fast_partial_schedule(current_time: int, permutation_task: ndarray[tuple[Any, ...], dtype[int64]], modes_array: ndarray[tuple[Any, ...], dtype[int64]], completed_task_indicator: ndarray[tuple[Any, ...], dtype[int64]], completed_task_times: ndarray[tuple[Any, ...], dtype[int64]], scheduled_task: ndarray[tuple[Any, ...], dtype[int64]], consumption_array: ndarray[tuple[Any, ...], dtype[int64]], duration_array: ndarray[tuple[Any, ...], dtype[int64]], predecessors: ndarray[tuple[Any, ...], dtype[int64]], successors: ndarray[tuple[Any, ...], dtype[int64]], horizon: int, ressource_available: ndarray[tuple[Any, ...], dtype[int64]], ressource_renewable: ndarray[tuple[Any, ...], dtype[bool]], minimum_starting_time_array: ndarray[tuple[Any, ...], dtype[int64]]) → tuple[dict[int, tuple[int, int]], bool][source]
discrete_optimization.rcpsp.fast_function.sgs_fast_partial_schedule_incomplete_permutation_tasks(current_time: int, permutation_task: ndarray[tuple[Any, ...], dtype[int64]], modes_array: ndarray[tuple[Any, ...], dtype[int64]], completed_task_indicator: ndarray[tuple[Any, ...], dtype[int64]], completed_task_times: ndarray[tuple[Any, ...], dtype[int64]], scheduled_task: ndarray[tuple[Any, ...], dtype[int64]], consumption_array: ndarray[tuple[Any, ...], dtype[int64]], duration_array: ndarray[tuple[Any, ...], dtype[int64]], predecessors: ndarray[tuple[Any, ...], dtype[int64]], successors: ndarray[tuple[Any, ...], dtype[int64]], horizon: int, ressource_available: ndarray[tuple[Any, ...], dtype[int64]], ressource_renewable: ndarray[tuple[Any, ...], dtype[bool]], minimum_starting_time_array: ndarray[tuple[Any, ...], dtype[int64]]) → tuple[dict[int, tuple[int, int]], bool][source]
discrete_optimization.rcpsp.fast_function.sgs_fast_partial_schedule_preemptive(current_time: int, permutation_task: ndarray[tuple[Any, ...], dtype[int64]], modes_array: ndarray[tuple[Any, ...], dtype[int64]], completed_task_indicator: ndarray[tuple[Any, ...], dtype[int64]], partial_schedule_starts: ndarray[tuple[Any, ...], dtype[int64]], partial_schedule_ends: ndarray[tuple[Any, ...], dtype[int64]], preemptive_tag: ndarray[tuple[Any, ...], dtype[bool]], consumption_array: ndarray[tuple[Any, ...], dtype[int64]], duration_array: ndarray[tuple[Any, ...], dtype[int64]], predecessors: ndarray[tuple[Any, ...], dtype[int64]], successors: ndarray[tuple[Any, ...], dtype[int64]], horizon: int, ressource_available: ndarray[tuple[Any, ...], dtype[int64]], ressource_renewable: ndarray[tuple[Any, ...], dtype[bool]], minimum_starting_time_array: ndarray[tuple[Any, ...], dtype[int64]]) → tuple[dict[int, ndarray[tuple[Any, ...], dtype[int64]]], dict[int, ndarray[tuple[Any, ...], dtype[int64]]], bool][source]
discrete_optimization.rcpsp.fast_function.sgs_fast_partial_schedule_preemptive_minduration(current_time: int, permutation_task: ndarray[tuple[Any, ...], dtype[int64]], modes_array: ndarray[tuple[Any, ...], dtype[int64]], completed_task_indicator: ndarray[tuple[Any, ...], dtype[int64]], partial_schedule_starts: ndarray[tuple[Any, ...], dtype[int64]], partial_schedule_ends: ndarray[tuple[Any, ...], dtype[int64]], preemptive_tag: ndarray[tuple[Any, ...], dtype[bool]], consumption_array: ndarray[tuple[Any, ...], dtype[int64]], duration_array: ndarray[tuple[Any, ...], dtype[int64]], predecessors: ndarray[tuple[Any, ...], dtype[int64]], successors: ndarray[tuple[Any, ...], dtype[int64]], horizon: int, ressource_available: ndarray[tuple[Any, ...], dtype[int64]], ressource_renewable: ndarray[tuple[Any, ...], dtype[bool]], min_duration_preemptive_bool: ndarray[tuple[Any, ...], dtype[bool]], min_duration_preemptive: ndarray[tuple[Any, ...], dtype[int64]]) → tuple[dict[int, ndarray[tuple[Any, ...], dtype[int64]]], dict[int, ndarray[tuple[Any, ...], dtype[int64]]], bool][source]
discrete_optimization.rcpsp.fast_function.sgs_fast_preemptive(permutation_task: ndarray[tuple[Any, ...], dtype[int64]], modes_array: ndarray[tuple[Any, ...], dtype[int64]], consumption_array: ndarray[tuple[Any, ...], dtype[int64]], duration_array: ndarray[tuple[Any, ...], dtype[int64]], preemptive_tag: ndarray[tuple[Any, ...], dtype[bool]], predecessors: ndarray[tuple[Any, ...], dtype[int64]], successors: ndarray[tuple[Any, ...], dtype[int64]], horizon: int, ressource_available: ndarray[tuple[Any, ...], dtype[int64]], ressource_renewable: ndarray[tuple[Any, ...], dtype[bool]], minimum_starting_time_array: ndarray[tuple[Any, ...], dtype[int64]]) → tuple[dict[int, ndarray[tuple[Any, ...], dtype[int64]]], dict[int, ndarray[tuple[Any, ...], dtype[int64]]], bool][source]
discrete_optimization.rcpsp.fast_function.sgs_fast_preemptive_minduration(permutation_task: ndarray[tuple[Any, ...], dtype[int64]], modes_array: ndarray[tuple[Any, ...], dtype[int64]], consumption_array: ndarray[tuple[Any, ...], dtype[int64]], duration_array: ndarray[tuple[Any, ...], dtype[int64]], preemptive_tag: ndarray[tuple[Any, ...], dtype[bool]], predecessors: ndarray[tuple[Any, ...], dtype[int64]], successors: ndarray[tuple[Any, ...], dtype[int64]], horizon: int, ressource_available: ndarray[tuple[Any, ...], dtype[int64]], ressource_renewable: ndarray[tuple[Any, ...], dtype[bool]], min_duration_preemptive_bool: ndarray[tuple[Any, ...], dtype[bool]], min_duration_preemptive: ndarray[tuple[Any, ...], dtype[int64]]) → tuple[dict[int, ndarray[tuple[Any, ...], dtype[int64]]], dict[int, ndarray[tuple[Any, ...], dtype[int64]]], bool][source]
discrete_optimization.rcpsp.fast_function.sgs_fast_preemptive_some_special_constraints(permutation_task: ndarray[tuple[Any, ...], dtype[int64]], modes_array: ndarray[tuple[Any, ...], dtype[int64]], consumption_array: ndarray[tuple[Any, ...], dtype[int64]], duration_array: ndarray[tuple[Any, ...], dtype[int64]], preemptive_tag: ndarray[tuple[Any, ...], dtype[bool]], predecessors: ndarray[tuple[Any, ...], dtype[int64]], successors: ndarray[tuple[Any, ...], dtype[int64]], start_at_end_plus_offset: ndarray[tuple[Any, ...], dtype[int64]], start_after_nunit: ndarray[tuple[Any, ...], dtype[int64]], minimum_starting_time_array: ndarray[tuple[Any, ...], dtype[int64]], horizon: int, ressource_available: ndarray[tuple[Any, ...], dtype[int64]], ressource_renewable: ndarray[tuple[Any, ...], dtype[bool]]) → tuple[dict[int, ndarray[tuple[Any, ...], dtype[int64]]], dict[int, ndarray[tuple[Any, ...], dtype[int64]]], bool][source]

discrete_optimization.rcpsp.parser module

discrete_optimization.rcpsp.parser.get_data_available(data_folder: str | None = None, data_home: str | None = None) → list[str][source]

Get datasets available for rcpsp.

Params:
data_folder: folder where datasets for rcpsp whould be find.

If None, we look in “rcpsp” subdirectory of data_home.

data_home: root directory for all datasets. Is None, set by

default to “~/discrete_optimization_data “

discrete_optimization.rcpsp.parser.parse_file(file_path: str) → RcpspProblem[source]
discrete_optimization.rcpsp.parser.parse_patterson(input_data: str) → RcpspProblem[source]
discrete_optimization.rcpsp.parser.parse_psplib(input_data: str) → RcpspProblem[source]

discrete_optimization.rcpsp.problem module

class discrete_optimization.rcpsp.problem.RcpspProblem(resources: dict[str, int | list[int]], non_renewable_resources: list[str], mode_details: dict[Hashable, dict[int, dict[str, int]]], successors: dict[Hashable, list[Hashable]], horizon: int, horizon_multiplier: int = 1, tasks_list: list[Hashable] | None = None, source_task: Hashable | None = None, sink_task: Hashable | None = None, name_task: dict[Hashable, str] | None = None, calendar_details: dict[str, list[list[int]]] | None = None, special_constraints: SpecialConstraintsDescription | None = None, relax_the_start_at_end: bool = True, fixed_permutation: list[int] | None = None, fixed_modes: list[int] | None = None, **kwargs: Any)[source]

Bases: MultimodeProblem[Hashable], SchedulingProblem[Hashable], PrecedenceProblem[Hashable]

Main class for RCPSP problem.

resources

mapping: name -> number of resources available, either at each time (-if an integer), or time step by time step (if a list of integers).

Type:

dict[str, Union[int, list[int]

non_renewable_resources

list of resources (specified by name) that are non-renewable

Type:

list[str]

mode_details

task -> (mode number -> resource needs and duration) Ex: >>> mode_details = { “task1” : … { # task1 details … 1: { “duration”: 2, “R1”: 1, “R2”: 0, “R3”: 4 }, # task1, mode 1: duration = 2, resource needs: R1=1, R3=4 … 2: { “duration”: 10, “R1”: 0, “R2”: 2, “R3”: 0 }, # task1, mode 1: duration = 10, resource needs: R2=2 … }, … # [other tasks] … }

Type:

dict[Hashable, dict[int, dict[str, int]]]

successors

successors in the precedence graph of each task

Type:

dict[Hashable, list[Hashable]]

horizon

max number of time steps allowed

Type:

int

horizon_multiplier

not used

tasks_list

list of tasks. Must correspond to mode_details. If not given will be equal to list(mode_details).

Type:

list[Hashable]

source_task

the id for the source task (i.e. the root in the precedence graph, all other tasks must follow it). If not given and tasks_list is made of integers, the lowest one

Type:

Hashable

sink_task

the id for the source task (i.e. the last task in the precedence graph, follows all other tasks). If not given and tasks_list is made of integers, the highest one.

Type:

Hashable

tasks_list_non_dummy

tasks list excluding source and sink. Same order as tasks_list

Type:

list[Hashable]

name_task

mapping task ids to task names. Default to id -> str(id)

Type:

dict[Hashable, str]

n_jobs

numer of tasks (i.e. len(tasks_list))

Type:

int

n_jobs_non_dummy

number of tasks excluding dummy activities Start (0) and End (n). (i.e. n_jobs - 2)

Type:

int

index_task

mapping task id to index among all tasks (from 0 to n_jobs)

Type:

dict[Hashable, int]

index_task_non_dummy

mapping task id to index among all non-dummy tasks (from 0 to n_jobs_non_dummy)

Type:

dict[Hashable, int]

special_constraints

special additional constraints

Type:

SpecialConstraintsDescription

do_special_constraints

True if they are special constraints to take into account

Type:

bool

relax_the_start_at_end

relax some conditions, only if do_special_constraints.

Type:

bool

fixed_permutation

To be used by solutions specifying only rcpsp_modes, e.g. in alternating genetic algorithms, that are updating only one solution attribute at a time. See RcpspSolution for more details.

Type:

Optional[list[int]]

fixed_modes

To be used by solutions specifying only the rpsp_permutation, e.g. in alternating genetic algorithms, that are updating only one solution attribute at a time. See RcpspSolution for more details.

Type:

Optional[list[int]]

Example

Initializing a simple RCPSP problem with 4 tasks, 2 resources, and precedence constraints, then generating a baseline schedule.

>>> from discrete_optimization.rcpsp.problem import RcpspProblem
>>> from discrete_optimization.rcpsp.solution import RcpspSolution
>>> # Define resources
>>> resources = {"R1": 5, "R2": 2}
>>> non_renewable_resources = []  # Empty if all resources replenish when a task ends
>>> # Define mode details
>>> # Format: { task_id: { mode_id: { "duration": int, "resource_name": amount } } }
>>> mode_details = {
...     "source": {1: {"duration": 0, "R1": 0, "R2": 0}},  # Source (Dummy)
...     "task-1": {1: {"duration": 4, "R1": 2, "R2": 2}},
...     "task-2": {1: {"duration": 3, "R1": 3, "R2": 0}},
...     "task-3": {1: {"duration": 5, "R1": 2, "R2": 1}},
...     "task-4": {1: {"duration": 2, "R1": 1, "R2": 1}},
...     "sink": {1: {"duration": 0, "R1": 0, "R2": 0}},  # Sink (Dummy)
... }
>>> # Define precedence graph
>>> successors = {
...     "source": ["task-1", "task-2"],  # First tasks: 1 and 2
...     "task-1": ["task-3"],            # Task 1 must finish before 3
...     "task-2": ["task-4"],            # Task 2 must finish before 4
...     "task-3": ["sink"],              # Task 3 leads to Sink
...     "task-4": ["sink"],              # Task 4 leads to Sink
...     "sink": [],                       # Sink has no successors
... }
>>> # Initialize the Problem
>>> problem = RcpspProblem(
...     resources=resources,
...     non_renewable_resources=non_renewable_resources,
...     mode_details=mode_details,
...     successors=successors,
...     horizon=20, # Maximum time allowed for the project
...     source_task="source",
...     sink_task="sink"
... )
>>> # Get a dummy solution: use serial sgs on trivial permutation
>>> # See RcpspSolution docstring for more information.
>>> sol = problem.get_dummy_solution()
>>> print(sol.rcpsp_schedule)
{'source': {'start_time': 0, 'end_time': 0}, 'task-1': {'start_time': 0, 'end_time': 4}, 'task-2': {'start_time': 0, 'end_time': 3}, 'task-3': {'start_time': 4, 'end_time': 9}, 'task-4': {'start_time': 4, 'end_time': 6}, 'sink': {'start_time': 9, 'end_time': 9}}
>>> # index_task vs index_task_non_dummy
>>> problem.index_task  # index among all tasks: task-1 -> 1
{'source': 0, 'task-1': 1, 'task-2': 2, 'task-3': 3, 'task-4': 4, 'sink': 5}
>>> problem.index_task_non_dummy  # index among only non-dummy tasks: task-1 -> 0
{'task-1': 0, 'task-2': 1, 'task-3': 2, 'task-4': 3}
>>> # Convert permutation of tasks into a list opf non-dummy indices
>>> problem.convert_task_ids_to_permutation(["task-3", "task-2", "task-4", "task-1"])
[2, 1, 3, 0]
build_mode_array(rcpsp_modes_from_solution: list[int]) → list[int][source]
build_mode_dict(rcpsp_modes_from_solution: list[int]) → dict[Hashable, int][source]
compute_graph(compute_predecessors: bool = False) → Graph[source]
compute_resource_consumption(rcpsp_sol: RcpspSolution) → ndarray[tuple[Any, ...], dtype[int64]][source]
convert_task_ids_to_permutation(tasks: list[Hashable]) → list[int][source]

Convert list of task ids into a permutation in proper format for RcpspSolution.

  • remove source and sink if present

  • use the indices among non-dummy tasks

Parameters:

tasks

Returns:

copy() → RcpspProblem[source]
evaluate(variable: RcpspSolution) → dict[str, float][source]

Evaluate a given solution object for the given problem.

This method should return a dictionnary of KPI, that can be then used for mono or multiobjective optimization.

Parameters:

variable (Solution) – the Solution object to evaluate.

Returns: dictionnary of float kpi for the solution.

evaluate_function(rcpsp_sol: RcpspSolution) → tuple[int, float, int][source]
evaluate_mobj(variable: RcpspSolution) → TupleFitness[source]

Default implementation of multiobjective evaluation.

It consists in flattening the evaluate() function and put in an array. User should probably custom this to be more efficient.

Parameters:

variable (Solution) – the Solution object to evaluate.

Returns (TupleFitness): a flattened tuple fitness object representing the multi-objective criteria.

evaluate_mobj_from_dict(dict_values: dict[str, float]) → TupleFitness[source]

Return an multiobjective fitness from a dictionnary of kpi (output of evaluate function).

It consists in flattening the evaluate() function and put in an array. User should probably custom this to be more efficient.

Parameters:

dict_values – output of evaluate() function

Returns (TupleFitness): a flattened tuple fitness object representing the multi-objective criteria.

get_attribute_register() → EncodingRegister[source]

Returns how the Solution should be encoded.

Useful to find automatically available mutations for local search. Used by genetic algorithms Ga and Nsga.

This needs only to be implemented in child classes when GA or LS solvers are to be used.

Returns (EncodingRegister): content of the encoding of the solution

get_dummy_solution() → RcpspSolution[source]

Apply serial sgs on the trivial task permutation.

See RcpspSolution about details on conversion permutation -> schedule.

Returns:

get_last_tasks() → list[Hashable][source]

Get a sublist of tasks that are candidate to be the last one scheduled.

Default to all tasks.

get_makespan_upper_bound() → int[source]

Get a upper bound on global makespan.

get_max_resource_capacity(res: str) → int[source]
get_objective_register() → ObjectiveRegister[source]

Returns the objective definition.

Returns (ObjectiveRegister): object defining the objective criteria.

get_precedence_constraints() → dict[Hashable, Iterable[Hashable]][source]

Map each task to the tasks that need to be performed after it.

get_resource_availability_array(res: str) → list[int][source]
get_resource_available(res: str, time: int) → int[source]
get_resource_names() → list[str][source]
get_solution_type() → type[Solution][source]

Returns the class implementation of a Solution.

Returns (class): class object of the given Problem.

get_task_modes(task: Hashable) → set[int][source]

Retrieve mode found for given task.

Parameters:

task

Returns:

includes_special_constraint() → bool[source]
is_multiskill() → bool[source]
is_preemptive() → bool[source]
is_rcpsp_multimode() → bool[source]
is_varying_resource() → bool[source]
plot_ressource_view(rcpsp_sol: RcpspSolution) → None[source]
return_index_task(task: Hashable, offset: int = 0) → int[source]
satisfy(variable: RcpspSolution) → bool[source]

Computes if a solution satisfies or not the constraints of the problem.

Parameters:

variable – the Solution object to check satisfability

Returns (bool): boolean true if the constraints are fulfilled, false elsewhere.

set_fixed_attributes(attribute_name: str, solution: RcpspSolution) → None[source]

Fix some solution attribute.

Useful when applying successively GA on different attribute of the solution, fixing the others.

Should be implemented at least for attributes described by attribute_register.

Parameters:

  • attribute_name – an attribute name

  • solution

Returns:

set_fixed_modes(fixed_modes: list[int]) → None[source]
set_fixed_permutation(fixed_permutation: list[int]) → None[source]
sgs: ScheduleGenerationScheme
property tasks_list: list[Hashable]

List of all tasks to schedule or allocate to.

update_functions() → None[source]
class discrete_optimization.rcpsp.problem.ScheduleGenerationScheme(*values)[source]

Bases: Enum

PARALLEL_SGS = 1
SERIAL_SGS = 0
discrete_optimization.rcpsp.problem.check_pair_mode_constraint(solution: RcpspSolution, pair_mode_constraint: PairModeConstraint)[source]
discrete_optimization.rcpsp.problem.check_solution_with_special_constraints(problem: RcpspProblem, solution: RcpspSolution, relax_the_start_at_end: bool = True) → bool[source]
discrete_optimization.rcpsp.problem.compute_constraints_details(solution: RcpspSolution, constraints: SpecialConstraintsDescription) → list[tuple[str, Hashable, Hashable, int | None, int | None, int]][source]
discrete_optimization.rcpsp.problem.compute_details_mode_constraint(solution: RcpspSolution, pair_mode_constraint: PairModeConstraint)[source]
discrete_optimization.rcpsp.problem.create_np_data_and_jit_functions(rcpsp_problem: RcpspProblem) → tuple[Callable[[...], tuple[dict[int, tuple[int, int]], bool]], Callable[[...], tuple[dict[int, tuple[int, int]], bool]], Callable[[...], float]][source]
discrete_optimization.rcpsp.problem.evaluate_constraints(solution: RcpspSolution, constraints: SpecialConstraintsDescription) → int[source]

discrete_optimization.rcpsp.problem_preemptive module

class discrete_optimization.rcpsp.problem_preemptive.PartialPreemptiveRcpspSolution(task_mode: dict[int, int] = None, start_times: dict[int, int] = None, end_times: dict[int, int] = None, partial_permutation: list[int] = None, list_partial_order: list[list[int]] = None, start_together: list[tuple[int, int]] = None, start_at_end: list[tuple[int, int]] = None, start_at_end_plus_offset: list[tuple[int, int, int]] = None, start_after_nunit: list[tuple[int, int, int]] = None, disjunctive_tasks: list[tuple[int, int]] = None, start_times_window: dict[Hashable, tuple[int, int]] = None, end_times_window: dict[Hashable, tuple[int, int]] = None)[source]

Bases: object

class discrete_optimization.rcpsp.problem_preemptive.PreemptiveRcpspProblem(resources: dict[str, int] | dict[str, list[int]], non_renewable_resources: list[str], mode_details: dict[Hashable, dict[str | int, dict[str, int]]], successors: dict[int | str, list[str | int]], horizon, horizon_multiplier=1, tasks_list: list[int | str] = None, source_task=None, sink_task=None, preemptive_indicator: dict[Hashable, bool] = None, duration_subtask: dict[Hashable, tuple[bool, int]] = None, name_task: dict[int, str] = None)[source]

Bases: Problem

build_mode_array(rcpsp_modes_from_solution)[source]
build_mode_dict(rcpsp_modes_from_solution)[source]
can_be_preempted(task)[source]
compute_graph() → Graph[source]
compute_resource_consumption(rcpsp_sol: PreemptiveRcpspSolution)[source]
copy()[source]
copy_with_multiplier(multiplier=0.5)[source]
evaluate(variable: PreemptiveRcpspSolution) → dict[str, float][source]

Evaluate a given solution object for the given problem.

This method should return a dictionnary of KPI, that can be then used for mono or multiobjective optimization.

Parameters:

variable (Solution) – the Solution object to evaluate.

Returns: dictionnary of float kpi for the solution.

evaluate_function(rcpsp_sol: PreemptiveRcpspSolution)[source]
evaluate_mobj(variable: PreemptiveRcpspSolution)[source]

Default implementation of multiobjective evaluation.

It consists in flattening the evaluate() function and put in an array. User should probably custom this to be more efficient.

Parameters:

variable (Solution) – the Solution object to evaluate.

Returns (TupleFitness): a flattened tuple fitness object representing the multi-objective criteria.

evaluate_mobj_from_dict(dict_values: dict[str, float]) → TupleFitness[source]

Return an multiobjective fitness from a dictionnary of kpi (output of evaluate function).

It consists in flattening the evaluate() function and put in an array. User should probably custom this to be more efficient.

Parameters:

dict_values – output of evaluate() function

Returns (TupleFitness): a flattened tuple fitness object representing the multi-objective criteria.

get_attribute_register() → EncodingRegister[source]

Returns how the Solution should be encoded.

Useful to find automatically available mutations for local search. Used by genetic algorithms Ga and Nsga.

This needs only to be implemented in child classes when GA or LS solvers are to be used.

Returns (EncodingRegister): content of the encoding of the solution

get_dummy_solution()[source]

Create a trivial solution for the problem.

Should satisfy the problem ideally. Does not exist for all kind of problems.

get_max_resource_capacity(res)[source]
get_modes_dict(rcpsp_solution: PreemptiveRcpspSolution)[source]
get_objective_register() → ObjectiveRegister[source]

Returns the objective definition.

Returns (ObjectiveRegister): object defining the objective criteria.

get_resource_availability_array(res)[source]
get_resource_available(res, time)[source]
get_resource_names()[source]
get_solution_type() → type[Solution][source]

Returns the class implementation of a Solution.

Returns (class): class object of the given Problem.

is_duration_minimum_preemption()[source]
is_multiskill()[source]
is_preemptive()[source]
is_rcpsp_multimode()[source]
is_varying_resource()[source]
mode_details: dict[Hashable, dict[int, dict[str, int]]]
n_jobs: int
non_renewable_resources: list[str]
plot_ressource_view(rcpsp_sol: PreemptiveRcpspSolution)[source]
resources: dict[str, int] | dict[str, list[int]]
return_index_task(task, offset=0)[source]
satisfy(variable: PreemptiveRcpspSolution) → bool[source]

Computes if a solution satisfies or not the constraints of the problem.

Parameters:

variable – the Solution object to check satisfability

Returns (bool): boolean true if the constraints are fulfilled, false elsewhere.

sgs: ScheduleGenerationScheme
successors: dict[int, list[int]]
update_function()[source]
class discrete_optimization.rcpsp.problem_preemptive.PreemptiveRcpspSolution(problem: PreemptiveRcpspProblem, rcpsp_permutation: list[int] | ndarray | None = None, rcpsp_schedule: dict[Hashable, dict[str, list[int]]] | None = None, rcpsp_modes: list[int] | None = None, rcpsp_schedule_feasible: bool | None = None, standardised_permutation: list[int] | ndarray | None = None)[source]

Bases: Solution

change_problem(new_problem: Problem)[source]

If relevant to the optimisation problem, change the underlying problem instance for the solution.

This method can be used to evaluate a solution for different instance of problems. It should be implemented in child classes when caching subresults depending on the problem.

Parameters:

new_problem (Problem) – another problem instance from which the solution can be evaluated

Returns: None

compute_mean_resource_reserve(fast=True)[source]
copy()[source]

Deep copy of the solution.

The copy() function should return a new object containing the same input as the current object, that respects the following expected behaviour: -y = x.copy() -if do some inplace change of y, the changes are not done in x.

Returns: a new object from which you can manipulate attributes without changing the original object.

generate_permutation_from_schedule()[source]
generate_schedule_from_permutation_serial_sgs(do_fast=True)[source]
generate_schedule_from_permutation_serial_sgs_2(current_t=0, completed_tasks: set[Hashable] | None = None, partial_schedule: dict[Hashable, dict[str, list[int]]] | None = None, do_fast=True)[source]
get_active_time(task)[source]
get_end_time(task)[source]
get_end_times_list(task)[source]
get_max_end_time()[source]
get_max_preempted()[source]
get_min_duration_subtask()[source]
get_nb_task_preemption()[source]
get_number_of_part(task)[source]
get_start_time(task)[source]
get_start_times_list(task)[source]
get_task_preempted()[source]
lazy_copy()[source]

This function should return a new object but possibly with mutable attributes from the original objects.

A typical use of lazy copy is in evolutionary algorithms or genetic algorithm where the use of local move don’t need to do a possibly costly deepcopy.

Returns (Solution): copy (possibly shallow) of the Solution

problem: PreemptiveRcpspProblem
rcpsp_modes: list[int]
rcpsp_permutation: list[int] | ndarray
rcpsp_schedule: dict[Hashable, dict[str, list[int]]]
standardised_permutation: list[int] | ndarray
total_number_of_cut()[source]
class discrete_optimization.rcpsp.problem_preemptive.ScheduleGenerationScheme(*values)[source]

Bases: Enum

PARALLEL_SGS = 1
SERIAL_SGS = 0
discrete_optimization.rcpsp.problem_preemptive.compute_mean_resource_reserve(solution: PreemptiveRcpspSolution, rcpsp_problem: PreemptiveRcpspProblem)[source]
discrete_optimization.rcpsp.problem_preemptive.compute_resource(solution: PreemptiveRcpspSolution, rcpsp_problem: PreemptiveRcpspProblem)[source]
discrete_optimization.rcpsp.problem_preemptive.create_np_data_and_jit_functions(rcpsp_problem: PreemptiveRcpspProblem)[source]
discrete_optimization.rcpsp.problem_preemptive.generate_schedule_from_permutation_serial_sgs(solution: PreemptiveRcpspSolution, rcpsp_problem: PreemptiveRcpspProblem)[source]
discrete_optimization.rcpsp.problem_preemptive.generate_schedule_from_permutation_serial_sgs_partial_schedule(solution: PreemptiveRcpspSolution, rcpsp_problem: PreemptiveRcpspProblem, partial_schedule: dict[Hashable, dict[str, list[int]]], current_t: int, completed_tasks: set[Hashable]) → tuple[dict[int, dict[str, list[int]]], bool][source]
discrete_optimization.rcpsp.problem_preemptive.get_rcpsp_problemp_preemptive(rcpsp_problem)[source]
discrete_optimization.rcpsp.problem_preemptive.permutation_do_to_permutation_sgs_fast(rcpsp_problem: PreemptiveRcpspProblem, permutation_do: Iterable[int]) → ndarray[tuple[Any, ...], dtype[int64]][source]
discrete_optimization.rcpsp.problem_preemptive.tree()[source]

discrete_optimization.rcpsp.problem_robust module

class discrete_optimization.rcpsp.problem_robust.AggregRcpspProblem(list_problem: Sequence[RcpspProblem], method_aggregating: MethodAggregating)[source]

Bases: RobustProblem, RcpspProblem

evaluate(variable: RcpspSolution) → dict[str, float][source]

Aggregated evaluate function.

Parameters:

variable (Solution) – Solution to evaluate on the different scenarios.

Returns (dict[str,float]): aggregated kpi on different scenarios.

evaluate_from_encoding(int_vector: list[int], encoding_name: str) → dict[str, float][source]

Evaluate a solution represented by its encoding.

Necessary to apply a genetic algorithm.

Parameters:

  • int_vector – solution attribute seen as a list of integer

  • encoding_name – name of the solution attribute used. Should be an entry of self.get_attribute_register().

Returns:

get_dummy_solution() → RcpspSolution[source]

Create a trivial solution for the problem.

Should satisfy the problem ideally. Does not exist for all kind of problems.

get_unique_rcpsp_problem() → RcpspProblem[source]
list_problem: Sequence[RcpspProblem]
class discrete_optimization.rcpsp.problem_robust.MethodBaseRobustification(*values)[source]

Bases: Enum

AVERAGE = 0
BEST_CASE = 2
PERCENTILE = 3
SAMPLE = 4
WORST_CASE = 1
class discrete_optimization.rcpsp.problem_robust.MethodRobustification(method_base: MethodBaseRobustification, percentile: float = 50)[source]

Bases: object

method_base: MethodBaseRobustification
percentile: float
class discrete_optimization.rcpsp.problem_robust.UncertainRcpspProblem(base_rcpsp_problem: RcpspProblem, poisson_laws: dict[Hashable, dict[int, dict[str, tuple[int, int, int]]]], uniform_law: bool = True)[source]

Bases: object

create_rcpsp_problem(method_robustification: MethodRobustification) → RcpspProblem[source]
probas: dict[Hashable, dict[int, dict[str, dict[str, Any]]]]
discrete_optimization.rcpsp.problem_robust.create_poisson_laws(base_rcpsp_problem: RcpspProblem, range_around_mean_resource: int = 1, range_around_mean_duration: int = 3, do_uncertain_resource: bool = True, do_uncertain_duration: bool = True) → dict[Hashable, dict[int, dict[str, tuple[int, int, int]]]][source]
discrete_optimization.rcpsp.problem_robust.create_poisson_laws_duration(rcpsp_problem: RcpspProblem, range_around_mean: int = 3) → dict[Hashable, dict[int, dict[str, tuple[int, int, int]]]][source]
discrete_optimization.rcpsp.problem_robust.create_poisson_laws_resource(rcpsp_problem: RcpspProblem, range_around_mean: int = 1) → dict[Hashable, dict[int, dict[str, tuple[int, int, int]]]][source]
discrete_optimization.rcpsp.problem_robust.tree() → dict[Any, Any][source]

discrete_optimization.rcpsp.problem_specialized_constraints module

class discrete_optimization.rcpsp.problem_specialized_constraints.SpecialConstraintsPreemptiveRcpspProblem(resources: dict[str, int] | dict[str, list[int]], non_renewable_resources: list[str], mode_details: dict[Hashable, dict[str | int, dict[str, int]]], successors: dict[int | str, list[str | int]], horizon, special_constraints: SpecialConstraintsDescription = None, preemptive_indicator: dict[Hashable, bool] = None, relax_the_start_at_end: bool = True, tasks_list: list[int | str] = None, source_task=None, sink_task=None, name_task: dict[int, str] = None, **kwargs)[source]

Bases: PreemptiveRcpspProblem

copy()[source]
evaluate(variable: PreemptiveRcpspSolution) → dict[str, float][source]

Evaluate a given solution object for the given problem.

This method should return a dictionnary of KPI, that can be then used for mono or multiobjective optimization.

Parameters:

variable (Solution) – the Solution object to evaluate.

Returns: dictionnary of float kpi for the solution.

evaluate_function(rcpsp_sol: PreemptiveRcpspSolution)[source]
get_dummy_solution(random_perm: bool = False)[source]

Create a trivial solution for the problem.

Should satisfy the problem ideally. Does not exist for all kind of problems.

get_objective_register() → ObjectiveRegister[source]

Returns the objective definition.

Returns (ObjectiveRegister): object defining the objective criteria.

get_solution_type() → type[Solution][source]

Returns the class implementation of a Solution.

Returns (class): class object of the given Problem.

has_special_constraints()[source]
is_preemptive()[source]
lazy_copy()[source]
satisfy(variable: PreemptiveRcpspSolution)[source]

Computes if a solution satisfies or not the constraints of the problem.

Parameters:

variable – the Solution object to check satisfability

Returns (bool): boolean true if the constraints are fulfilled, false elsewhere.

update_function()[source]
update_functions()[source]
class discrete_optimization.rcpsp.problem_specialized_constraints.SpecialPreemptiveRcpspSolution(problem: PreemptiveRcpspProblem, rcpsp_permutation: list[int] | ndarray | None = None, rcpsp_schedule: dict[Hashable, dict[str, list[int]]] | None = None, rcpsp_modes: list[int] | None = None, rcpsp_schedule_feasible: bool | None = None, standardised_permutation: list[int] | ndarray | None = None)[source]

Bases: PreemptiveRcpspSolution

change_problem(new_problem: Problem)[source]

If relevant to the optimisation problem, change the underlying problem instance for the solution.

This method can be used to evaluate a solution for different instance of problems. It should be implemented in child classes when caching subresults depending on the problem.

Parameters:

new_problem (Problem) – another problem instance from which the solution can be evaluated

Returns: None

copy()[source]

Deep copy of the solution.

The copy() function should return a new object containing the same input as the current object, that respects the following expected behaviour: -y = x.copy() -if do some inplace change of y, the changes are not done in x.

Returns: a new object from which you can manipulate attributes without changing the original object.

generate_schedule_from_permutation_serial_sgs(do_fast=True)[source]
generate_schedule_from_permutation_serial_sgs_2(current_t: int = 0, completed_tasks: set[Hashable] | None = None, partial_schedule: dict[Hashable, dict[str, list[int]]] | None = None, do_fast: bool = True)[source]
lazy_copy()[source]

This function should return a new object but possibly with mutable attributes from the original objects.

A typical use of lazy copy is in evolutionary algorithms or genetic algorithm where the use of local move don’t need to do a possibly costly deepcopy.

Returns (Solution): copy (possibly shallow) of the Solution

discrete_optimization.rcpsp.problem_specialized_constraints.check_solution(problem: SpecialConstraintsPreemptiveRcpspProblem, solution: SpecialPreemptiveRcpspSolution | RcpspSolution | PreemptiveRcpspSolution, relax_the_start_at_end: bool = True)[source]
discrete_optimization.rcpsp.problem_specialized_constraints.compute_constraints_details(solution: RcpspSolution | PreemptiveRcpspSolution, constraints: SpecialConstraintsDescription)[source]
discrete_optimization.rcpsp.problem_specialized_constraints.create_np_data_and_jit_functions(rcpsp_problem: SpecialConstraintsPreemptiveRcpspProblem)[source]
discrete_optimization.rcpsp.problem_specialized_constraints.evaluate_constraints(solution: RcpspSolution | PreemptiveRcpspSolution, constraints: SpecialConstraintsDescription)[source]
discrete_optimization.rcpsp.problem_specialized_constraints.generate_schedule_from_permutation_serial_sgs_partial_schedule_preempptive(solution, rcpsp_problem, partial_schedule: dict[Hashable, dict[str, list[int]]], current_t, completed_tasks)[source]
discrete_optimization.rcpsp.problem_specialized_constraints.generate_schedule_from_permutation_serial_sgs_preemptive(solution, rcpsp_problem: SpecialConstraintsPreemptiveRcpspProblem)[source]

discrete_optimization.rcpsp.sgs_without_array module

class discrete_optimization.rcpsp.sgs_without_array.SgsWithoutArray(rcpsp_problem: RcpspProblem)[source]

Bases: object

static add_event_delta(sdict: SortedDict, time_start: int, delta: int, time_end: int) → None[source]
static add_event_delta_in_absolute(sdict: SortedDict, time_start: int, delta: int, time_end: int, liberate: bool = True) → None[source]
static create_absolute_dict(vector: ArrayLike) → SortedDict[source]
static create_delta_dict(vector: ArrayLike) → SortedDict[source]
generate_schedule_from_permutation_serial_sgs(solution: RcpspSolution, rcpsp_problem: RcpspProblem) → tuple[dict[Hashable, dict[str, int]], bool, dict[str, SortedDict]][source]
static get_available_from_absolute(sdict: SortedDict, time: int) → int[source]
static get_available_from_delta(sdict: SortedDict, time: int) → int[source]

discrete_optimization.rcpsp.solution module

class discrete_optimization.rcpsp.solution.PartialSolution(task_mode: dict[int, int] | None = None, start_times: dict[int, int] | None = None, end_times: dict[int, int] | None = None, partial_permutation: list[int] | None = None, list_partial_order: list[list[int]] | None = None, start_together: list[tuple[int, int]] | None = None, start_at_end: list[tuple[int, int]] | None = None, start_at_end_plus_offset: list[tuple[int, int, int]] | None = None, start_after_nunit: list[tuple[int, int, int]] | None = None, disjunctive_tasks: list[tuple[int, int]] | None = None, start_times_window: dict[Hashable, tuple[int, int]] | None = None, end_times_window: dict[Hashable, tuple[int, int]] | None = None)[source]

Bases: object

class discrete_optimization.rcpsp.solution.RcpspSolution(problem: RcpspProblem, rcpsp_permutation: list[int] | None = None, rcpsp_schedule: dict[Hashable, dict[str, int]] | None = None, rcpsp_modes: list[int] | None = None, rcpsp_schedule_feasible: bool = True, standardised_permutation: list[int] | None = None, fast: bool = True)[source]

Bases: SchedulingSolution[Hashable], MultimodeSolution[Hashable]

Solution to RcpspProblem problems.

problem

RCPSP problem for which this is a solution

Type:

RcpspProblem

rcpsp_permutation

Tasks permutation, i.e. list of tasks indices among non-dummy tasks (i.e. excluding source and sink tasks). See below how it is related to the schedule. If given and schedule not given, used to reconstruct the schedule. if not given, deduced from schedule. If not given and schedule not given, it is set to problem.fixed_permutation (useful for alternating genetic algorithms)

rcpsp_schedule

Tasks schedule. ( task -> “start_time” or “end_time” -> time) If given used to construct standardised_permutation. If given and rcpsp_permutation not given, used to construct rcpsp_permutation. If given and rcpsp_permutation given, no consistency check. If not given, deduced from rcpsp_permutation if possible. If not possible, rcpsp_schedule_feasible set to False and rcpsp_schedule set to a partially filled schedule. In case of Aggreg_RcpspModel, kept empty.

rcpsp_modes

Mode used for each task. Same order as problem.tasks_list_non_dummy. If not given we use problem.fixed_modes if existing, else 1 for each task.

rcpsp_schedule_feasible

True if a schedule can be deduced from permutation. False if it leads to incoherency preventing a schedule generation. Recomputed when schedule is (re)computed from permutation.

standardised_permutation

Permutation deduced uniquely from schedule. If given, not recomputed. Can be different from rcpsp_permutation as different permutations can lead to same schedule.

fast

boolean indicating if we use the fast functions to generate schedule from permutation.

## More about conversion permutation <-> schedule:

### Schedule -> permutation Take the list of tasks indices among non-dummy tasks (i.e. excluding source and sink tasks), sorted by their starting times.

### Permutation -> schedule: serial-SGS (Schedule Generation Scheme) - Initialization:

  • the source task is set at time 0

  • the resource availability is set to the initial calendar given by the problem

  • Loop: the schedule is filled task by task, in the order given by the permutation, and taking into account

    • the precedence constraints: a task starts after the end of the tasks before it in the precedence graph

    • the resource constraints: during the task, the needed resource must be available

    When a new task has been inserted, the availability of each resource is updated accordingly, i.e. we remove the necessary resources for the task (according to the chosen mode set by rcpsp_modes) for the task duration, and after if the resource is non-renewable.

  • When hitting an impossibility (e.g. inserting a task after one of its successor, or when the resources are not enough available until the problem horizon) the permutation is set infeasible (so rcpsp_schedule_feasible is set to False).

See “Compute a dummy solution for RCPSP” in the tutorial “notebooks/RCPSP tutorials/RCPSP-1 Introduction.ipynb” for more details.

Example

This example demonstrates how a permutation of non-dummy tasks is transformed into a schedule, adhering to precedence and resource constraints.

>>> from discrete_optimization.rcpsp.problem import RcpspProblem
>>> from discrete_optimization.rcpsp.solution import RcpspSolution
>>> # Define resources
>>> resources = {"R1": 5, "R2": 2}
>>> non_renewable_resources = []  # Empty if all resources replenish when a task ends
>>> # Define mode details
>>> # Format: { task_id: { mode_id: { "duration": int, "resource_name": amount } } }
>>> mode_details = {
...     "source": {1: {"duration": 0, "R1": 0, "R2": 0}},  # Source (Dummy)
...     "task-1": {1: {"duration": 4, "R1": 2, "R2": 2}},
...     "task-2": {1: {"duration": 3, "R1": 3, "R2": 0}},
...     "task-3": {1: {"duration": 5, "R1": 2, "R2": 1}},
...     "task-4": {1: {"duration": 2, "R1": 1, "R2": 1}},
...     "sink": {1: {"duration": 0, "R1": 0, "R2": 0}},  # Sink (Dummy)
... }
>>> # Define precedence graph
>>> successors = {
...     "source": ["task-1", "task-2"],  # First tasks: 1 and 2
...     "task-1": ["task-3"],            # Task 1 must finish before 3
...     "task-2": ["task-4"],            # Task 2 must finish before 4
...     "task-3": ["sink"],              # Task 3 leads to Sink
...     "task-4": ["sink"],              # Task 4 leads to Sink
...     "sink": [],                       # Sink has no successors
... }
>>> # Initialize the Problem
>>> problem = RcpspProblem(
...     resources=resources,
...     non_renewable_resources=non_renewable_resources,
...     mode_details=mode_details,
...     successors=successors,
...     horizon=20, # Maximum time allowed for the project
...     source_task="source",
...     sink_task="sink"
... )
>>> # Create solution
>>> # Convert list of task ids into a licit permutation (using indices among non-dummy tasks)
>>> rcpsp_permutation = problem.convert_task_ids_to_permutation(["task-2", "task-1", "task-4", "task-3"])
>>> print(rcpsp_permutation)
[1, 0, 3, 2]
>>> solution = RcpspSolution(
...     problem=problem,
...     rcpsp_permutation=rcpsp_permutation,
...     rcpsp_modes=[1, 1, 1, 1]  # same mode for each task
... )
>>> # The Serial SGS calculates the start/end times according:
>>> print(solution.rcpsp_schedule["task-2"])  # Task 2
{'start_time': 0, 'end_time': 3}
>>> print(solution.rcpsp_schedule["task-1"])  # Task 1 follows Task 2 in permutation
{'start_time': 0, 'end_time': 4}
>>> print(solution.rcpsp_schedule["task-4"])  # Task 4 can start after Task 2, but only when ressources are available (thus after task 1)
{'start_time': 4, 'end_time': 6}
>>> print(solution.rcpsp_schedule["task-3"])  # Task 3 can start after Task 1 (and ressources are available)
{'start_time': 4, 'end_time': 9}
change_problem(new_problem: RcpspProblem) → None[source]

If relevant to the optimisation problem, change the underlying problem instance for the solution.

This method can be used to evaluate a solution for different instance of problems. It should be implemented in child classes when caching subresults depending on the problem.

Parameters:

new_problem (Problem) – another problem instance from which the solution can be evaluated

Returns: None

compute_mean_resource_reserve(fast: bool = True) → float[source]
copy() → RcpspSolution[source]

Deep copy of the solution.

The copy() function should return a new object containing the same input as the current object, that respects the following expected behaviour: -y = x.copy() -if do some inplace change of y, the changes are not done in x.

Returns: a new object from which you can manipulate attributes without changing the original object.

generate_permutation_from_schedule() → list[int][source]
generate_schedule_from_permutation_serial_sgs(do_fast: bool = True) → None[source]
generate_schedule_from_permutation_serial_sgs_2(current_t: int = 0, completed_tasks: dict[Hashable, TaskDetails] | None = None, scheduled_tasks_start_times: dict[Hashable, int] | None = None, do_fast: bool = True) → None[source]
get_active_time(task: Hashable) → list[int][source]
get_end_time(task: Hashable) → int[source]
get_end_times_list(task: Hashable) → list[int][source]
get_max_end_time() → int[source]
get_mode(task: Hashable) → int[source]

Retrieve mode found for given task.

Parameters:

task

Returns:

get_start_time(task: Hashable) → int[source]
get_start_times_list(task: Hashable) → list[int][source]
lazy_copy() → RcpspSolution[source]

This function should return a new object but possibly with mutable attributes from the original objects.

A typical use of lazy copy is in evolutionary algorithms or genetic algorithm where the use of local move don’t need to do a possibly costly deepcopy.

Returns (Solution): copy (possibly shallow) of the Solution

problem: RcpspProblem
class discrete_optimization.rcpsp.solution.TaskDetails(start: int, end: int)[source]

Bases: object

discrete_optimization.rcpsp.solution.compute_mean_resource_reserve(solution: RcpspSolution, rcpsp_problem: RcpspProblem) → float[source]
discrete_optimization.rcpsp.solution.generate_schedule_from_permutation_serial_sgs(solution: RcpspSolution, rcpsp_problem: RcpspProblem) → tuple[dict[Hashable, dict[str, int]], bool][source]
discrete_optimization.rcpsp.solution.generate_schedule_from_permutation_serial_sgs_partial_schedule(solution: RcpspSolution, rcpsp_problem: RcpspProblem, current_t: int, completed_tasks: dict[Hashable, TaskDetails], scheduled_tasks_start_times: dict[Hashable, int]) → tuple[dict[Hashable, dict[str, int]], bool][source]
discrete_optimization.rcpsp.solution.generate_schedule_from_permutation_serial_sgs_partial_schedule_specialized_constraints(solution: RcpspSolution, rcpsp_problem: RcpspProblem, current_t: int, completed_tasks: dict[Hashable, TaskDetails], scheduled_tasks_start_times: dict[Hashable, int]) → tuple[dict[Hashable, dict[str, int]], bool][source]
discrete_optimization.rcpsp.solution.generate_schedule_from_permutation_serial_sgs_special_constraints(solution: RcpspSolution, rcpsp_problem: RcpspProblem) → tuple[dict[Hashable, dict[str, int]], bool][source]
discrete_optimization.rcpsp.solution.permutation_do_to_permutation_sgs_fast(rcpsp_problem: RcpspProblem, permutation_do: Iterable[int]) → npt.NDArray[np.int_][source]

discrete_optimization.rcpsp.solvers_map module

discrete_optimization.rcpsp.solvers_map.get_solver_default_arguments(method: type[RcpspSolver] | type[GenericRcpspSolver]) → dict[str, Any][source]
discrete_optimization.rcpsp.solvers_map.look_for_solver(domain: RcpspProblem | PreemptiveRcpspProblem | SpecialConstraintsPreemptiveRcpspProblem) → list[type[RcpspSolver] | type[GenericRcpspSolver]][source]
discrete_optimization.rcpsp.solvers_map.look_for_solver_class(class_domain: type[RcpspProblem | PreemptiveRcpspProblem | SpecialConstraintsPreemptiveRcpspProblem]) → list[type[RcpspSolver] | type[GenericRcpspSolver]][source]
discrete_optimization.rcpsp.solvers_map.return_solver(method: type[RcpspSolver] | type[GenericRcpspSolver], problem: RcpspProblem | PreemptiveRcpspProblem | SpecialConstraintsPreemptiveRcpspProblem, **kwargs: Any) → RcpspSolver | GenericRcpspSolver[source]
discrete_optimization.rcpsp.solvers_map.solve(method: type[RcpspSolver] | type[GenericRcpspSolver], problem: RcpspProblem | PreemptiveRcpspProblem | SpecialConstraintsPreemptiveRcpspProblem, **kwargs: Any) → ResultStorage[source]
discrete_optimization.rcpsp.solvers_map.solve_return_solver(method: type[RcpspSolver] | type[GenericRcpspSolver], problem: RcpspProblem | PreemptiveRcpspProblem | SpecialConstraintsPreemptiveRcpspProblem, **kwargs: Any) → tuple[ResultStorage, RcpspSolver | GenericRcpspSolver][source]

discrete_optimization.rcpsp.special_constraints module

class discrete_optimization.rcpsp.special_constraints.PairModeConstraint(allowed_mode_assignment: dict[tuple[Hashable, Hashable], set[tuple[int, int]]] | None = None, same_score_mode: set[tuple[Hashable, Hashable]] | None = None, score_mode: dict[tuple[Hashable, int], int] = None)[source]

Bases: object

This data structure exists to express the following constraints : if allowed_mode_assignment is not None : for each key of the dictionary (activity_1, activity_2), the allowed modes assignment to those 2 activity are listed in allowed_mode_assignment[(activity_1, activity_2)] This can be used to express that a given resource has to do both activities.

Another way of expressing the constraint is : -each interesting tuple (task,mode) is given an int score. (score_mode) -We get a set of pairs of activities (same_score_mode) : indicating that the score_mode of the 2 activity should be equal.

class discrete_optimization.rcpsp.special_constraints.SpecialConstraintsDescription(task_mode: dict[Hashable, int] | None = None, start_times: dict[Hashable, int] | None = None, end_times: dict[Hashable, int] | None = None, start_times_window: dict[Hashable, tuple[int | None, int | None]] | None = None, end_times_window: dict[Hashable, tuple[int | None, int | None]] | None = None, partial_permutation: list[Hashable] | None = None, list_partial_order: list[list[Hashable]] | None = None, start_together: list[tuple[Hashable, Hashable]] | None = None, start_at_end: list[tuple[Hashable, Hashable]] | None = None, start_at_end_plus_offset: list[tuple[Hashable, Hashable, int]] | None = None, start_after_nunit: list[tuple[Hashable, Hashable, int]] | None = None, disjunctive_tasks: list[tuple[Hashable, Hashable]] | None = None, pair_mode_constraint: PairModeConstraint | None = None)[source]

Bases: object

discrete_optimization.rcpsp.transform_problem module

discrete_optimization.rcpsp.transform_problem.from_rcpsp_problem(rcpsp_problem: RcpspProblem, constraints: SpecialConstraintsDescription, preemptive=False) → RcpspProblem[source]

discrete_optimization.rcpsp.utils module

discrete_optimization.rcpsp.utils.all_diff_start_time(rcpsp_sols: tuple[RcpspSolution, RcpspSolution]) → dict[Hashable, int][source]
discrete_optimization.rcpsp.utils.compute_graph_rcpsp(rcpsp_problem: RcpspProblem) → Graph[source]
discrete_optimization.rcpsp.utils.compute_nice_resource_consumption(rcpsp_problem: RcpspProblem, rcpsp_sol: RcpspSolution, list_resources: list[str] | None = None) → tuple[dict[int, npt.NDArray[np.int_]], dict[int, npt.NDArray[np.int_]]][source]
discrete_optimization.rcpsp.utils.compute_resource_consumption(rcpsp_problem: RcpspProblem, rcpsp_sol: RcpspSolution, list_resources: list[str] | None = None, future_view: bool = True) → tuple[npt.NDArray[np.int_], npt.NDArray[np.int_]][source]
discrete_optimization.rcpsp.utils.compute_schedule_per_resource_individual(rcpsp_problem: RcpspProblem, rcpsp_sol: RcpspSolution, resource_types_to_consider: list[str] | None = None) → dict[str, dict[str, Any]][source]
discrete_optimization.rcpsp.utils.create_fake_tasks(rcpsp_problem: RcpspProblem) → list[dict[str, int]][source]
discrete_optimization.rcpsp.utils.get_end_bounds_from_additional_constraint(rcpsp_problem: RcpspProblem, activity: Hashable) → tuple[int, int][source]
discrete_optimization.rcpsp.utils.get_max_time_solution(solution: PreemptiveRcpspSolution | RcpspSolution) → int[source]
discrete_optimization.rcpsp.utils.get_start_bounds_from_additional_constraint(rcpsp_problem: RcpspProblem, activity: Hashable) → tuple[int, int][source]
discrete_optimization.rcpsp.utils.get_tasks_ending_between_two_times(solution: PreemptiveRcpspSolution | RcpspSolution, time_1: int, time_2: int) → list[Hashable][source]
discrete_optimization.rcpsp.utils.intersect(i1: Sequence[int], i2: Sequence[int]) → list[int] | None[source]
discrete_optimization.rcpsp.utils.kendall_tau_similarity(rcpsp_sols: tuple[RcpspSolution, RcpspSolution]) → float[source]
discrete_optimization.rcpsp.utils.plot_resource_individual_gantt(rcpsp_problem: RcpspProblem, rcpsp_sol: RcpspSolution, resource_types_to_consider: list[str] | None = None, title_figure: str = '', x_lim: list[int] | None = None, fig: plt.Figure | None = None, ax: npt.NDArray[np.object_] | None = None, current_t: int | None = None) → plt.Figure[source]
discrete_optimization.rcpsp.utils.plot_ressource_view(rcpsp_problem: RcpspProblem, rcpsp_sol: RcpspSolution, list_resource: list[str] | None = None, title_figure: str = '', x_lim: list[int] | None = None, fig: plt.Figure | None = None, ax: npt.NDArray[np.object_] | None = None) → plt.Figure[source]
discrete_optimization.rcpsp.utils.plot_task_gantt(rcpsp_problem: RcpspProblem, rcpsp_sol: RcpspSolution, fig: plt.Figure | None = None, ax: plt.Axes | None = None, x_lim: list[int] | None = None, title: str | None = None) → plt.Figure[source]

discrete_optimization.rcpsp.utils_preemptive module

discrete_optimization.rcpsp.utils_preemptive.compute_nice_resource_consumption(rcpsp_problem: PreemptiveRcpspProblem, rcpsp_sol: PreemptiveRcpspSolution, list_resources: list[int | str] = None)[source]
discrete_optimization.rcpsp.utils_preemptive.compute_resource_consumption(rcpsp_problem: PreemptiveRcpspProblem, rcpsp_sol: PreemptiveRcpspSolution, list_resources: list[int | str] = None, future_view=True)[source]
discrete_optimization.rcpsp.utils_preemptive.compute_schedule_per_resource_individual(rcpsp_problem: PreemptiveRcpspProblem, rcpsp_sol: PreemptiveRcpspSolution, resource_types_to_consider: list[str] = None)[source]
discrete_optimization.rcpsp.utils_preemptive.plot_resource_individual_gantt(rcpsp_problem: PreemptiveRcpspProblem, rcpsp_sol: PreemptiveRcpspSolution, resource_types_to_consider: list[str] = None, title_figure='', x_lim=None, fig=None, ax=None, current_t=None)[source]
discrete_optimization.rcpsp.utils_preemptive.plot_ressource_view(rcpsp_problem: PreemptiveRcpspProblem, rcpsp_sol: PreemptiveRcpspSolution, list_resource: list[int | str] = None, title_figure='', x_lim=None, fig=None, ax=None)[source]
discrete_optimization.rcpsp.utils_preemptive.plot_task_gantt(rcpsp_problem: PreemptiveRcpspProblem, rcpsp_sol: PreemptiveRcpspSolution, fig=None, ax=None, x_lim=None, title=None, current_t=None)[source]

Module contents