discrete_optimization.rcpsp_multiskill package

Subpackages

Submodules

discrete_optimization.rcpsp_multiskill.fast_function_ms_rcpsp module

discrete_optimization.rcpsp_multiskill.fast_function_ms_rcpsp.sgs_fast_ms(permutation_task, priority_worker_per_task, modes_array, consumption_array, skills_needs, duration_array, predecessors, successors, horizon, ressource_available, ressource_renewable, worker_available, worker_skills, minimum_starting_time_array, one_unit_per_task: bool = True)[source]
discrete_optimization.rcpsp_multiskill.fast_function_ms_rcpsp.sgs_fast_ms_partial_schedule(permutation_task, priority_worker_per_task, modes_array, scheduled_task_indicator, scheduled_start_task_times, scheduled_end_task_times, worker_used, current_time, consumption_array, skills_needs, duration_array, predecessors, successors, horizon, ressource_available, ressource_renewable, worker_available, worker_skills, minimum_starting_time_array, one_unit_per_task: bool = True)[source]
discrete_optimization.rcpsp_multiskill.fast_function_ms_rcpsp.sgs_fast_ms_preemptive(permutation_task, priority_worker_per_task, modes_array, consumption_array, skills_needs, duration_array, preemptive_tag, predecessors, successors, horizon, ressource_available, ressource_renewable, worker_available, worker_skills, minimum_starting_time_array, is_releasable, one_unit_per_task: bool = True, consider_partial_preemptive: bool = False, strictly_disjunctive_subtasks: bool = False)[source]
discrete_optimization.rcpsp_multiskill.fast_function_ms_rcpsp.sgs_fast_ms_preemptive_partial_schedule(permutation_task, priority_worker_per_task, modes_array, scheduled_task_indicator, scheduled_start_task_times, scheduled_end_task_times, nb_subparts, worker_used, current_time, consumption_array, skills_needs, duration_array, preemptive_tag, predecessors, successors, horizon, ressource_available, ressource_renewable, worker_available, worker_skills, minimum_starting_time_array, is_releasable, one_unit_per_task: bool = True, consider_partial_preemptive: bool = False)[source]
discrete_optimization.rcpsp_multiskill.fast_function_ms_rcpsp.sgs_fast_ms_preemptive_some_special_constraints(permutation_task, priority_worker_per_task, modes_array, consumption_array, skills_needs, duration_array, preemptive_tag, predecessors, successors, start_at_end_plus_offset, start_after_nunit, horizon, ressource_available, ressource_renewable, worker_available, worker_skills, minimum_starting_time_array, is_releasable, one_unit_per_task: bool = True, consider_partial_preemptive: bool = False, strictly_disjunctive_subtasks: bool = False)[source]

discrete_optimization.rcpsp_multiskill.multiskill_to_rcpsp module

class discrete_optimization.rcpsp_multiskill.multiskill_to_rcpsp.MultiSkillToRcpsp(multiskill_model: MultiskillRcpspProblem)[source]

Bases: object

construct_rcpsp_by_worker_type(limit_number_of_mode_per_task: bool = True, max_number_of_mode: int | None = None, check_resource_compliance: bool = True, one_worker_type_per_task: bool = False)[source]
is_compatible(task_requirements: dict[str, int], ressource_availability: dict[str, array], duration_task, horizon)[source]

discrete_optimization.rcpsp_multiskill.parser_imopse module

discrete_optimization.rcpsp_multiskill.parser_imopse.get_data_available(data_folder: str | None = None, data_home: str | None = None)[source]

Get datasets available for rcpsp_multiskill.

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

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

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

default to “~/discrete_optimization_data “

discrete_optimization.rcpsp_multiskill.parser_imopse.parse_file(file_path, max_horizon=None, one_unit_per_task=True, preemptive=False) tuple[MultiskillRcpspProblem, dict][source]
discrete_optimization.rcpsp_multiskill.parser_imopse.parse_imopse(input_data, max_horizon=None, one_unit_per_task=True, preemptive=False)[source]

discrete_optimization.rcpsp_multiskill.parser_mslib module

discrete_optimization.rcpsp_multiskill.parser_mslib.get_data_available(data_folder: str | None = None, data_home: str | None = None)[source]

Get datasets available for knapsack. Params:

data_folder: folder where datasets for knapsack whould be find.

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

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

default to “~/discrete_optimization_data “

discrete_optimization.rcpsp_multiskill.parser_mslib.parse_file(file_path, skill_level_version: bool = True)[source]

discrete_optimization.rcpsp_multiskill.parser_mspsp module

discrete_optimization.rcpsp_multiskill.parser_mspsp.get_data_available(data_folder: str | None = None, data_home: str | None = None)[source]
discrete_optimization.rcpsp_multiskill.parser_mspsp.parse_file(file_path) MultiskillRcpspProblem[source]

discrete_optimization.rcpsp_multiskill.problem module

class discrete_optimization.rcpsp_multiskill.problem.Employee(dict_skill: dict[str, SkillDetail], calendar_employee: list[bool], salary: float = 0.0)[source]

Bases: object

calendar_employee: list[bool]
copy()[source]
dict_skill: dict[str, SkillDetail]
get_non_zero_skills()[source]
get_skill_level(s)[source]
to_json(with_calendar: bool = True)[source]
class discrete_optimization.rcpsp_multiskill.problem.MultiskillRcpspProblem(skills_set: set[str], resources_set: set[str], non_renewable_resources: set[str], resources_availability: dict[str, list[int]], employees: dict[Hashable, Employee], mode_details: dict[Hashable, dict[int, dict[str, int]]], successors: dict[Hashable, list[Hashable]], horizon, employees_availability: list[int] | None = None, tasks_list: list[Hashable] | None = None, employees_list: list[Hashable] | None = None, horizon_multiplier=1, sink_task: Hashable | None = None, source_task: Hashable | None = None, one_unit_per_task_max: bool = False, preemptive: bool = False, preemptive_indicator: dict[Hashable, bool] | None = None, special_constraints: SpecialConstraintsDescription | None = None, partial_preemption_data: dict[Hashable, dict[int, dict[str, bool]]] | None = None, always_releasable_resources: set[str] | None = None, never_releasable_resources: set[str] | None = None, resource_blocking_data: list[tuple[list[Hashable], set[str]]] | None = None, strictly_disjunctive_subtasks: bool = True)[source]

Bases: Problem

build_mode_dict(rcpsp_modes_from_solution)[source]
build_multimode_rcpsp_calendar_representative()[source]
compute_graph() Graph[source]
copy()[source]
employees: dict[Hashable, Employee]
employees_availability: list[int]
evaluate(rcpsp_sol: MultiskillRcpspSolution) 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.

abstract evaluate_from_encoding(int_vector, encoding_name)[source]
evaluate_function(rcpsp_sol: MultiskillRcpspSolution)[source]
evaluate_mobj(rcpsp_sol: MultiskillRcpspSolution)[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: MultiskillRcpspSolution)[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]
includes_special_constraint()[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_non_dummy: int
non_renewable_resources: set[str]
partial_preemption_data: dict[Hashable, dict[int, dict[str, bool]]]
resource_blocking_data: list[tuple[list[Hashable], set[str]]]
resources_availability: dict[str, list[int]]
resources_set: set[str]
return_index_task(task, offset=0)[source]
satisfy(variable: Solution) 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.

satisfy_classic(rcpsp_sol: MultiskillRcpspSolution) bool[source]
satisfy_preemptive(rcpsp_sol: PreemptiveMultiskillRcpspSolution) bool[source]
sgs: ScheduleGenerationScheme
skills_set: set[str]
strictly_disjunctive_subtasks: bool
successors: dict[Hashable, list[Hashable]]
to_variant_model()[source]
update_function()[source]
update_functions()[source]
class discrete_optimization.rcpsp_multiskill.problem.MultiskillRcpspSolution(problem: Problem, modes: dict[Hashable, int], schedule: dict[Hashable, dict[str, int | list[int]]], employee_usage: dict[Hashable, dict[Hashable, set[str]]])[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

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.

employee_used(task)[source]
get_active_time(task)[source]
get_end_time(task)[source]
get_end_times_list(task)[source]
get_max_end_time()[source]
get_number_of_part(task)[source]
get_start_time(task)[source]
get_start_times_list(task)[source]
class discrete_optimization.rcpsp_multiskill.problem.PreemptiveMultiskillRcpspSolution(problem: Problem, modes: dict[Hashable, int], schedule: dict[Hashable, dict[str, list[int]]], employee_usage: dict[Hashable, list[dict[Hashable, set[str]]]])[source]

Bases: MultiskillRcpspSolution

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.

employee_used(task)[source]
get_active_time(task)[source]
get_end_time(task)[source]
get_end_times_list(task)[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]
total_number_of_cut()[source]
class discrete_optimization.rcpsp_multiskill.problem.ScheduleGenerationScheme(value)[source]

Bases: Enum

An enumeration.

PARALLEL_SGS = 1
SERIAL_SGS = 0
class discrete_optimization.rcpsp_multiskill.problem.SkillDetail(skill_value: float, efficiency_ratio: float, experience: float)[source]

Bases: object

copy()[source]
efficiency_ratio: float
experience: float
skill_value: float
class discrete_optimization.rcpsp_multiskill.problem.TaskDetails(start, end, resource_units_used: list[int])[source]

Bases: object

class discrete_optimization.rcpsp_multiskill.problem.TaskDetailsPreemptive(starts: list[int], ends: list[int], resource_units_used: list[list[Hashable]])[source]

Bases: object

class discrete_optimization.rcpsp_multiskill.problem.VariantMultiskillRcpspProblem(skills_set: set[str], resources_set: set[str], non_renewable_resources: set[str], resources_availability: dict[str, list[int]], employees: dict[Hashable, Employee], employees_availability: list[int], mode_details: dict[Hashable, dict[int, dict[str, int]]], successors: dict[Hashable, list[Hashable]], horizon, tasks_list: list[Hashable] | None = None, employees_list: list[Hashable] | None = None, horizon_multiplier=1, sink_task: Hashable | None = None, source_task: Hashable | None = None, one_unit_per_task_max: bool = False, preemptive: bool = False, preemptive_indicator: dict[Hashable, bool] | None = None, special_constraints: SpecialConstraintsDescription | None = None, partial_preemption_data: dict[Hashable, dict[int, dict[str, bool]]] | None = None, always_releasable_resources: set[str] | None = None, never_releasable_resources: set[str] | None = None, resource_blocking_data: list[tuple[list[Hashable], set[str]]] | None = None, strictly_disjunctive_subtasks: bool = True)[source]

Bases: MultiskillRcpspProblem

convert_fixed_priority_worker_per_task_from_permutation(permutation)[source]
evaluate_from_encoding(int_vector, encoding_name)[source]
evaluate_function(rcpsp_sol: VariantMultiskillRcpspSolution)[source]
get_attribute_register() EncodingRegister[source]

Returns how the Solution should be encoded.

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

get_dummy_solution(preemptive: bool | None = None)[source]
get_solution_type() type[Solution][source]

Returns the class implementation of a Solution.

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

set_fixed_attributes(encoding_str: str, sol: VariantMultiskillRcpspSolution)[source]
set_fixed_modes(fixed_modes)[source]
set_fixed_priority_worker_per_task(fixed_priority_worker_per_task)[source]
set_fixed_priority_worker_per_task_from_permutation(permutation)[source]
set_fixed_task_permutation(fixed_permutation)[source]
class discrete_optimization.rcpsp_multiskill.problem.VariantMultiskillRcpspSolution(problem: Problem, modes_vector: list[int] | None = None, modes_vector_from0: list[int] | None = None, priority_list_task: list[int] | None = None, priority_worker_per_task: list[list[Hashable]] | None = None, modes: dict[int, int] | None = None, schedule: dict[int, dict[str, int]] | None = None, employee_usage: dict[int, dict[int, set[str]]] | None = None, fast: bool = True)[source]

Bases: MultiskillRcpspSolution

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.

do_recompute(fast=True)[source]
run_sgs_partial(current_t, completed_tasks: dict[Hashable, TaskDetails], scheduled_tasks_start_times: dict[Hashable, TaskDetails], fast=True)[source]
update_infos_from_numba_output(rcpsp_schedule, skills_usage, unfeasible_non_renewable_resources)[source]
class discrete_optimization.rcpsp_multiskill.problem.VariantPreemptiveMultiskillRcpspSolution(problem: Problem, modes_vector: list[int] | None = None, modes_vector_from0: list[int] | None = None, priority_list_task: list[int] | None = None, priority_worker_per_task: list[list[Hashable]] | None = None, modes: dict[int, int] | None = None, schedule: dict[Hashable, dict[str, list[int]]] | None = None, employee_usage: dict[Hashable, list[dict[Hashable, set[str]]]] | None = None, fast: bool = True)[source]

Bases: PreemptiveMultiskillRcpspSolution

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.

do_recompute(fast=True)[source]
run_sgs_partial(current_t, completed_tasks: dict[Hashable, TaskDetailsPreemptive], scheduled_tasks_start_times: dict[Hashable, TaskDetailsPreemptive], fast: bool = True)[source]
update_from_numba_output(starts_dict, ends_dict, skills_usage, unfeasible_non_renewable_resources)[source]
discrete_optimization.rcpsp_multiskill.problem.build_partial_vectors(problem: MultiskillRcpspProblem, completed_tasks: dict[Hashable, TaskDetails], scheduled_tasks_start_times: dict[Hashable, TaskDetails])[source]
discrete_optimization.rcpsp_multiskill.problem.build_partial_vectors_preemptive(problem: MultiskillRcpspProblem, completed_tasks: dict[Hashable, TaskDetailsPreemptive], scheduled_tasks_start_times: dict[Hashable, TaskDetailsPreemptive])[source]
discrete_optimization.rcpsp_multiskill.problem.check_solution(problem: MultiskillRcpspProblem, solution: MultiskillRcpspSolution | PreemptiveMultiskillRcpspSolution, relax_the_start_at_end: bool = True)[source]
discrete_optimization.rcpsp_multiskill.problem.cluster_employees_to_resource_types(ms_rcpsp_problem: MultiskillRcpspProblem)[source]
discrete_optimization.rcpsp_multiskill.problem.compute_constraints_details(solution: MultiskillRcpspSolution | PreemptiveMultiskillRcpspSolution, constraints: SpecialConstraintsDescription)[source]
discrete_optimization.rcpsp_multiskill.problem.compute_discretize_calendar_skills(problem: MultiskillRcpspProblem) tuple[dict[str, list[dict]], dict[str, ndarray]][source]
discrete_optimization.rcpsp_multiskill.problem.compute_overskill(problem: MultiskillRcpspProblem, solution: PreemptiveMultiskillRcpspSolution)[source]
discrete_optimization.rcpsp_multiskill.problem.compute_ressource_array_preemptive(problem: MultiskillRcpspProblem, solution: MultiskillRcpspSolution | PreemptiveMultiskillRcpspSolution)[source]
discrete_optimization.rcpsp_multiskill.problem.compute_skills_calendar(problem: MultiskillRcpspProblem)[source]
discrete_optimization.rcpsp_multiskill.problem.compute_skills_missing_problem(problem: MultiskillRcpspProblem, solution: MultiskillRcpspSolution | PreemptiveMultiskillRcpspSolution)[source]
discrete_optimization.rcpsp_multiskill.problem.create_fake_tasks_multiskills(rcpsp_problem: MultiskillRcpspProblem | VariantMultiskillRcpspSolution)[source]
discrete_optimization.rcpsp_multiskill.problem.create_np_data_and_jit_functions(rcpsp_problem: MultiskillRcpspProblem | VariantMultiskillRcpspProblem)[source]
discrete_optimization.rcpsp_multiskill.problem.discretize_calendar_(capacity_calendar: ndarray)[source]

Compute decrease of capacity slots as a list of (interval, consumption)

discrete_optimization.rcpsp_multiskill.problem.employee_usage(solution: MultiskillRcpspSolution | PreemptiveMultiskillRcpspSolution, problem: MultiskillRcpspProblem)[source]
discrete_optimization.rcpsp_multiskill.problem.evaluate_constraints(solution: MultiskillRcpspSolution | PreemptiveMultiskillRcpspSolution, constraints: SpecialConstraintsDescription)[source]
discrete_optimization.rcpsp_multiskill.problem.intersect(i1, i2)[source]
discrete_optimization.rcpsp_multiskill.problem.permutation_do_to_permutation_sgs_fast(rcpsp_problem: MultiskillRcpspProblem, permutation_do)[source]
discrete_optimization.rcpsp_multiskill.problem.priority_worker_per_task_do_to_permutation_sgs_fast(rcpsp_problem: MultiskillRcpspProblem, priority_worker_per_task)[source]
discrete_optimization.rcpsp_multiskill.problem.schedule_solution_preemptive_to_variant(solution: PreemptiveMultiskillRcpspSolution)[source]
discrete_optimization.rcpsp_multiskill.problem.schedule_solution_to_variant(solution: MultiskillRcpspSolution)[source]
discrete_optimization.rcpsp_multiskill.problem.sgs_multi_skill(solution: VariantMultiskillRcpspSolution)[source]
discrete_optimization.rcpsp_multiskill.problem.sgs_multi_skill_partial_schedule(solution: VariantMultiskillRcpspSolution, current_t, completed_tasks: dict[Hashable, TaskDetails], scheduled_tasks_start_times: dict[Hashable, TaskDetails])[source]
discrete_optimization.rcpsp_multiskill.problem.sgs_multi_skill_preemptive(solution: VariantPreemptiveMultiskillRcpspSolution)[source]
discrete_optimization.rcpsp_multiskill.problem.sgs_multi_skill_preemptive_partial_schedule(solution: VariantPreemptiveMultiskillRcpspSolution, current_t, completed_tasks: dict[Hashable, TaskDetailsPreemptive], scheduled_tasks_start_times: dict[Hashable, TaskDetailsPreemptive])[source]
discrete_optimization.rcpsp_multiskill.problem.start_together_problem_description(solution: MultiskillRcpspSolution | PreemptiveMultiskillRcpspSolution, constraints: SpecialConstraintsDescription)[source]
discrete_optimization.rcpsp_multiskill.problem.tree()[source]

discrete_optimization.rcpsp_multiskill.solvers_map module

discrete_optimization.rcpsp_multiskill.solvers_map.look_for_solver(domain)[source]
discrete_optimization.rcpsp_multiskill.solvers_map.look_for_solver_class(class_domain)[source]
discrete_optimization.rcpsp_multiskill.solvers_map.return_solver(method, problem: MultiskillRcpspProblem, **args) ResultStorage[source]
discrete_optimization.rcpsp_multiskill.solvers_map.solve(method, problem: MultiskillRcpspProblem, **args) ResultStorage[source]

Module contents