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, 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: int, employees_availability: list[int] | None = None, tasks_list: list[Hashable] | None = None, employees_list: list[Hashable] | None = None, horizon_multiplier: int = 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, special_constraints: SpecialConstraintsDescription | None = None, partial_preemption_data: dict[Hashable, dict[int, dict[str, bool]]] = None, always_releasable_resources: set[str] = None, never_releasable_resources: set[str] = None, resource_blocking_data: list[tuple[list[Hashable], set[str]]] = None, strictly_disjunctive_subtasks: bool = True)[source]

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

build_mode_dict(rcpsp_modes_from_solution)[source]
build_multimode_rcpsp_calendar_representative()[source]
compute_graph() Graph[source]
copy()[source]
create_employee_task_compatibility() None[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.

evaluate_function(rcpsp_sol: MultiskillRcpspSolution)[source]
evaluate_mobj(variable: 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.

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_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)[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_precedence_constraints() dict[Hashable, list[Hashable]][source]

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

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_task_modes(task: Hashable) set[int][source]

Retrieve mode found for given task.

Parameters:

task

Returns:

includes_special_constraint()[source]
is_compatible_task_unary_resource(task: Hashable, unary_resource: Hashable) bool[source]

Should return False if the unary_resource can never be allocated to task.

This is only a hint used to reduce the number of variables or constraints generated.

Default to True, to be overriden in subclasses.

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]]
property tasks_list: list[Hashable]

List of all tasks to schedule or allocate to.

to_variant_model()[source]
property unary_resources_list: list[Hashable]

Get available unary resources for allocation.

Warning: This is inherited from AllocationProblem which is used here for the employees allocation. Thus “resource” means “employee” in that particular case.

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: SchedulingSolution[Hashable], MultimodeSolution[Hashable], AllocationSolution[Hashable, Hashable]

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_mode(task: Hashable) int[source]

Retrieve mode found for given task.

Parameters:

task

Returns:

get_number_of_part(task)[source]
get_start_time(task)[source]
get_start_times_list(task)[source]
is_allocated(task: Hashable, unary_resource: Hashable) bool[source]

Return the usage of the unary resource for the given task.

Parameters:
  • task

  • unary_resource

Returns:

problem: MultiskillRcpspProblem
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(*values)[source]

Bases: Enum

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, employees_list: list[Hashable] = 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, special_constraints: SpecialConstraintsDescription = None, partial_preemption_data: dict[Hashable, dict[int, dict[str, bool]]] = None, always_releasable_resources: set[str] = None, never_releasable_resources: set[str] = None, resource_blocking_data: list[tuple[list[Hashable], set[str]]] = None, strictly_disjunctive_subtasks: bool = True)[source]

Bases: MultiskillRcpspProblem

convert_fixed_priority_worker_per_task_from_permutation(permutation: list[int]) list[list[Hashable]][source]
convert_fixed_priority_worker_per_task_to_permutation(priority_worker_per_task: list[list[Hashable]]) list[int][source]
evaluate_function(rcpsp_sol: VariantMultiskillRcpspSolution)[source]
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(preemptive: bool | None = None)[source]

Create a trivial solution for the problem.

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

get_solution_type() type[Solution][source]

Returns the class implementation of a Solution.

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

set_fixed_attributes(attribute_name: str, solution: VariantMultiskillRcpspSolution) 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)[source]
set_fixed_priority_worker_per_task(fixed_priority_worker_per_task: list[list[Hashable]])[source]
set_fixed_priority_worker_per_task_from_permutation(permutation: list[int])[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, priority_worker_per_task_perm: list[int] | None = None, modes: dict[int, int] = None, schedule: dict[int, dict[str, int]] = None, employee_usage: dict[int, dict[int, set[str]]] = 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]
property priority_worker_per_task_perm: list[int]

Represent priority_worker_per_task as a big permutation.

problem: VariantMultiskillRcpspProblem
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, priority_worker_per_task_perm: list[int] | None = None, modes: dict[int, int] = None, schedule: dict[Hashable, dict[str, list[int]]] = None, employee_usage: dict[Hashable, list[dict[Hashable, set[str]]]] = 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]
property priority_worker_per_task_perm: list[int]

Represent priority_worker_per_task as a big permutation.

problem: VariantMultiskillRcpspProblem
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