discrete_optimization.rcpsp package

Subpackages

Submodules

discrete_optimization.rcpsp.fast_function module

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

discrete_optimization.rcpsp.mutation module

class discrete_optimization.rcpsp.mutation.DeadlineMutationRcpsp(problem: RcpspProblem | PreemptiveRcpspProblem | SpecialConstraintsPreemptiveRcpspProblem | MultiskillRcpspProblem | VariantMultiskillRcpspProblem, solution: PreemptiveRcpspSolution | RcpspSolution | MultiskillRcpspSolution | VariantMultiskillRcpspSolution | PreemptiveMultiskillRcpspSolution | VariantPreemptiveMultiskillRcpspSolution, attribute: str | None = None, nb_swap: int = 1)[source]

Bases: Mutation

static build(problem: Problem, solution: Solution, **kwargs: Any) → DeadlineMutationRcpsp[source]
mutate(solution: PreemptiveRcpspSolution | RcpspSolution | MultiskillRcpspSolution | VariantMultiskillRcpspSolution | PreemptiveMultiskillRcpspSolution | VariantPreemptiveMultiskillRcpspSolution) → tuple[PreemptiveRcpspSolution | RcpspSolution | MultiskillRcpspSolution | VariantMultiskillRcpspSolution | PreemptiveMultiskillRcpspSolution | VariantPreemptiveMultiskillRcpspSolution, LocalMove][source]
mutate_and_compute_obj(solution: Solution) → tuple[Solution, LocalMove, dict[str, float]][source]
class discrete_optimization.rcpsp.mutation.PermutationMutationRcpsp(problem: Problem, solution: Solution, other_mutation: Mutation)[source]

Bases: Mutation

static build(problem: ~discrete_optimization.generic_tools.do_problem.Problem, solution: ~discrete_optimization.generic_tools.do_problem.Solution, other_mutation: type[~discrete_optimization.generic_tools.do_mutation.Mutation] = <class 'discrete_optimization.generic_tools.mutations.permutation_mutations.PermutationShuffleMutation'>, **kwargs: ~typing.Any) → PermutationMutationRcpsp[source]
mutate(solution: RcpspSolution) → tuple[Solution, LocalMove][source]
mutate_and_compute_obj(solution: RcpspSolution) → tuple[Solution, LocalMove, dict[str, float]][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: Problem

resources
non_renewable_resources
mode_details
successors
horizon
horizon_multiplier
tasks_list
source_task
sink_task
name_task
n_jobs
Type:

int

n_jobs_non_dummy

excluding dummy activities Start (0) and End (n)

Type:

int

special_constraints
do_special_constraints
Type:

bool

relax_the_start_at_end

relax some conditions only if do_special_constraints

Type:

bool

fixed_permutation
Type:

Optional[list[int]]

fixed_modes
Type:

Optional[list[int]]

Parameters:

  • resources – {resource_name: number_of_resource}

  • non_renewable_resources – [resource_name3, resource_name4]

  • mode_details – {job_id: {mode_id: {resource_name1: number_of_resources_needed, resource_name2: …}} one key being “duration”

  • successors

  • horizon

  • horizon_multiplier

  • tasks_list – {task_id: list of successor task ids}

  • source_task

  • sink_task

  • name_task

  • special_constraints

  • relax_the_start_at_end

  • fixed_permutation

  • fixed_modes

  • **kwargs

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[Any, dtype[int64]][source]
copy() → RcpspProblem[source]
evaluate(rcpsp_sol: 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_from_encoding(int_vector: list[int], encoding_name: str) → dict[str, float][source]
evaluate_function(rcpsp_sol: RcpspSolution) → tuple[int, float, int][source]
evaluate_mobj(rcpsp_sol: 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.

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

get_dummy_solution() → RcpspSolution[source]
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_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_tasks_list() → list[Hashable][source]
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(rcpsp_sol: 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(encoding_str: str, sol: RcpspSolution) → None[source]
set_fixed_modes(fixed_modes: list[int]) → None[source]
set_fixed_permutation(fixed_permutation: list[int]) → None[source]
sgs: ScheduleGenerationScheme
update_functions() → None[source]
class discrete_optimization.rcpsp.problem.ScheduleGenerationScheme(value)[source]

Bases: Enum

An enumeration.

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 = 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.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 = None, source_task=None, sink_task=None, preemptive_indicator: dict[Hashable, bool] | None = None, duration_subtask: dict[Hashable, tuple[bool, int]] | None = None, name_task: dict[int, str] | None = 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(rcpsp_sol: 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_from_encoding(int_vector, encoding_name)[source]
evaluate_function(rcpsp_sol: PreemptiveRcpspSolution)[source]
evaluate_mobj(rcpsp_sol: 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.

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

get_dummy_solution()[source]
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.

get_tasks_list()[source]
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(rcpsp_sol: 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, rcpsp_permutation=None, rcpsp_schedule=None, rcpsp_modes=None, rcpsp_schedule_feasible=None, standardised_permutation=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.

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

rcpsp_modes: list[int]
rcpsp_permutation: list[int] | array
rcpsp_schedule: dict[Hashable, dict[str, list[int]]]
standardised_permutation: list[int] | array
total_number_of_cut()[source]
class discrete_optimization.rcpsp.problem_preemptive.ScheduleGenerationScheme(value)[source]

Bases: Enum

An enumeration.

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[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]
get_dummy_solution() → RcpspSolution[source]
get_unique_rcpsp_problem() → RcpspProblem[source]
list_problem: Sequence[RcpspProblem]
class discrete_optimization.rcpsp.problem_robust.MethodBaseRobustification(value)[source]

Bases: Enum

An enumeration.

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]
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 = None, preemptive_indicator: dict[Hashable, bool] | None = None, relax_the_start_at_end: bool = True, tasks_list: list[int | str] | None = None, source_task=None, sink_task=None, name_task: dict[int, str] | None = None, **kwargs)[source]

Bases: PreemptiveRcpspProblem

copy()[source]
evaluate(rcpsp_sol: 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_from_encoding(int_vector, encoding_name)[source]
evaluate_function(rcpsp_sol: PreemptiveRcpspSolution)[source]
get_dummy_solution(random_perm: bool = False)[source]
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(rcpsp_sol: 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, rcpsp_permutation=None, rcpsp_schedule=None, rcpsp_modes=None, rcpsp_schedule_feasible=None, standardised_permutation=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.

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: _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes]) → SortedDict[source]
static create_delta_dict(vector: _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes]) → 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: Solution

Solution to RcpspProblem problems.

problem

RCPSP problem for which this is a solution

rcpsp_permutation

Tasks permutation.

rcpsp_schedule

Tasks schedule. ( task -> “start_time” or “end_time” -> time) Potentially only partial if not feasible with given permutation, or for aggregated models.

rcpsp_modes

Mode used for each task. Same order as problem.tasks_list_non_dummy.

rcpsp_schedule_feasible

False if schedule generation from permutation failed, True else.

standardised_permutation

Permutation deduced uniquely from schedule. Can be different from rcpsp_permutation as different permutations can lead to same schedule.

fast

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

Parameters:

  • problem – RCPSP problem for which this is a solution

  • rcpsp_permutation – Tasks permutation. 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

  • 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 us the fast functions to generate schedule from permutation.

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.

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]
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

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 = 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 = None)[source]
discrete_optimization.rcpsp.utils_preemptive.compute_resource_consumption(rcpsp_problem: PreemptiveRcpspProblem, rcpsp_sol: PreemptiveRcpspSolution, list_resources: list[int | str] | None = 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 = None)[source]
discrete_optimization.rcpsp.utils_preemptive.plot_resource_individual_gantt(rcpsp_problem: PreemptiveRcpspProblem, rcpsp_sol: PreemptiveRcpspSolution, resource_types_to_consider: list[str] | None = 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 = 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