discrete_optimization.workforce.scheduling package

Subpackages

Submodules

discrete_optimization.workforce.scheduling.parser module

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

Get datasets available for tsp.

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

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

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

default to “~/discrete_optimization_data “

discrete_optimization.workforce.scheduling.parser.parse_json_to_problem(json_path: str) → AllocSchedulingProblem[source]
discrete_optimization.workforce.scheduling.parser.update_calendars_with_disruptions(list_drop_resource, teams: list[Hashable], calendar_team: dict[Hashable, list[tuple[int, int]]])[source]

discrete_optimization.workforce.scheduling.problem module

class discrete_optimization.workforce.scheduling.problem.AllocSchedulingProblem(team_names: list[Hashable], calendar_team: dict[Hashable, list[tuple[int, int]]], horizon: int, tasks_list: list[Hashable], tasks_data: dict[Hashable, TasksDescription], same_allocation: list[set[Hashable]], precedence_constraints: dict[Hashable, set[Hashable]], available_team_for_activity: dict[Hashable, set[Hashable]], start_window: dict[Hashable, tuple[int | None, int | None]], end_window: dict[Hashable, tuple[int | None, int | None]], original_start: dict[Hashable, int], original_end: dict[Hashable, int], resources_list: list[str] | None = None, resources_capacity: dict[str, int] | None = None, horizon_start_shift: int | None = 0, objective_handling: ObjectiveHandling = ObjectiveHandling.AGGREGATE)[source]

Bases: GenericSchedulingProblem[Hashable, Hashable, None, str, None], WithoutSkillProblem[Hashable, Hashable, str, Hashable], WithoutNonRenewableResourceProblem[Hashable], SinglemodeSchedulingProblem[Hashable]

compatible_teams_all_activity() → dict[Hashable, set[Hashable]][source]
compatible_teams_index_all_activity() → dict[int, set[int]][source]
compute_predecessors()[source]
compute_unavailability_calendar(team: Hashable) → list[tuple[int, int]][source]
evaluate(variable: Solution) → 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.

get_all_lb_ub() → list[tuple[int, int, int, int]][source]

Return a list of (lb_start, ub_start, lb_end, ub_end) for each task (lower/upper bound on start, lower/upper bound on end)

get_cumulative_resource_consumption(resource: Hashable | str, task: Hashable, mode: int) → int[source]

Get cumulative resource consumption of the task in the given mode

Parameters:
  • resource – cumulative resource

  • task

  • mode – not used for single mode problems

Returns:

the consumption for cumulative resources.

get_lb_end_window(task: Hashable) → int[source]
get_lb_start_window(task: Hashable) → int[source]
get_list_objective_computer() → list[ObjectiveComputer][source]
get_makespan_upper_bound() → int[source]

Get an upper bound on global makespan.

get_objective_register() → ObjectiveRegister[source]

Returns the objective definition.

Returns (ObjectiveRegister): object defining the objective criteria.

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

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

get_resource_availabilities(resource: Hashable | str) → list[tuple[int, int, int]][source]

Get availabilities intervals for a given resource

List of availability intervals of a resource. If the resource is not available, potentially no interval returned.

It is assumed that the intervals are disjunct though.

Parameters:

resource

Returns:

list of intervals of the form (start, end, value), which means from time start to time end, there are value of the resource available. NB: the start is included, the end is excluded (start <= t < end)

get_same_unary_allocation() → list[set[Hashable]][source]

Overridden from base AllocationProblem class

get_solution_type() → type[Solution][source]

Returns the class implementation of a Solution.

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

get_task_duration(task: Hashable) → int[source]

Get task duration.

Parameters:

task

Returns:

get_task_start_or_end_lower_bound(task: Hashable, start_or_end: StartOrEnd) → int[source]

Get a lower bound on start or end of a given task as specified by the problem.

For tighter computed bounds, see GenericSchedulingProblem.get_tight_task_start_or_end_lower_bound() and GenericSchedulingProblem.compute_task_bounds().

Default implementation: 0

Parameters:
  • task

  • start_or_end

Returns:

get_task_start_or_end_upper_bound(task: Hashable, start_or_end: StartOrEnd) → int[source]

Get an upper bound on start or end of a given task as specified by the problem.

For tighter computed bounds, see GenericSchedulingProblem.get_tight_task_start_or_end_upper_bound() and GenericSchedulingProblem.compute_task_bounds().

Default implementation: makespan upper bound

Parameters:
  • task

  • start_or_end

Returns:

get_ub_end_window(task: Hashable) → int[source]
get_ub_start_window(task: Hashable) → int[source]
get_unavailable_teams_per_activity() → dict[Hashable, set[Hashable]][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_optional(task: Hashable) → bool[source]

Whether a task is optional or not.

It means that the task can be ignored in the solution. If absent of the solution, it can also be removed from the constraints.

Default to no optional task.

property non_skill_cumulative_resources_list: list[str]

List of cumulative resources that are not skills.

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.

set_objective_handling(objective_handling: ObjectiveHandling)[source]
property tasks_list: list[Hashable]

List of all tasks to schedule or allocate to.

property unary_resources_list: list[Hashable]

Available unary resources.

It can correspond to employees (rcpsp-multiskill), teams (workforce-scheduling), or a mix of several types.

update_available_team_for_activity(available_team_for_activity: dict[Hashable, set[Hashable]])[source]

Override the available team attribute and update the rcpsp accordingly

update_problem()[source]

Method to call whenever available_team_for_activity or resources_capacity are updated.

class discrete_optimization.workforce.scheduling.problem.AllocSchedulingSolution(problem: AllocSchedulingProblem, schedule: ndarray, allocation: ndarray)[source]

Bases: GenericSchedulingSolution[Hashable, Hashable, None, str, None], WithoutSkillSolution[Hashable, Hashable, str, Hashable], WithoutNonRenewableResourceSolution[Hashable], SinglemodeSolution[Hashable]

copy() → Solution[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.

get_end_time(task: Hashable) → int | AbsentValue[source]

Get end time of the task

Hypothesis:

The returned time can have AbsentValue.ABSENT only if self.is_present(task) is False.

Parameters:

task

Returns:

get_start_time(task: Hashable) → int | AbsentValue[source]

Get start time of the task

Hypothesis:

The returned time can have AbsentValue.ABSENT only if self.is_present(task) is False.

Parameters:

task

Returns:

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:

is_present(task: Hashable) → bool[source]

Tell whether the task is present in the solution.

It can mean several thing: - scheduling problem: start and end of the task are defined - allocation problem: a ressource has been allocated to the task - multimode: a mode has been chosen for the task - or a mix of it

Sometimes a scheduling allocation problem will allow no resource allocation for present task, it has to be defined problem by problem.

For convenience, a default implementation is provided which assumes that all tasks are present. To be overriden in subclasses.

If and only if this method returns False, the mode, start, and end of the task can have the value AbsentValue.ABSENT.

If the method returns False, the task is removed from all constraints during checks.

problem: AllocSchedulingProblem
class discrete_optimization.workforce.scheduling.problem.TasksDescription(duration_task: int, resource_consumption: dict[str, int] = None)[source]

Bases: object

discrete_optimization.workforce.scheduling.problem.build_calendar_array_from_availability_slot(availability_slots: list[tuple[int, int]], horizon: int, value: int = 1)[source]
discrete_optimization.workforce.scheduling.problem.build_pair_mode_constraint(problem: AllocSchedulingProblem, rcpsp: RcpspProblem, use_score: bool = False)[source]
discrete_optimization.workforce.scheduling.problem.compute_stats_per_team(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution) → dict[int, float][source]
discrete_optimization.workforce.scheduling.problem.correct_schedule_avoid_overlap(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, init_min_starting_date_lb: bool = False)[source]
discrete_optimization.workforce.scheduling.problem.evaluate_solution(solution: AllocSchedulingSolution, problem: AllocSchedulingProblem) → dict[str, float][source]
discrete_optimization.workforce.scheduling.problem.export_scheduling_problem_json(problem: AllocSchedulingProblem) → dict[source]
discrete_optimization.workforce.scheduling.problem.full_satisfy(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, partial_solution: bool = False) → bool[source]
discrete_optimization.workforce.scheduling.problem.interval_inside(interval1: tuple[float, float], interval_container: tuple[float, float])[source]
discrete_optimization.workforce.scheduling.problem.intervals_do_not_overlap(interval1: tuple[float, float], interval2: tuple[float, float])[source]
discrete_optimization.workforce.scheduling.problem.realign_calendars(calendars_dict: dict[Hashable, list[tuple[int, int]]])[source]
discrete_optimization.workforce.scheduling.problem.satisfy_all_done(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, partial_solution: bool = False)[source]
discrete_optimization.workforce.scheduling.problem.satisfy_available_team(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, partial_solution: bool = False) → bool[source]

Partial solution = True means we ignore variable set to None when we check the constraint.

discrete_optimization.workforce.scheduling.problem.satisfy_calendar_resources(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, partial_solution: bool = False)[source]

Check constraints related to renewable calendar resources (teams + cumulative resources)

It checks: - team availability when allocated to a task - non overlap of tasks with same team allocated - cumulative resources availability for all tasks

Parameters:
  • problem

  • solution

  • partial_solution

Returns:

discrete_optimization.workforce.scheduling.problem.satisfy_calendars(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, partial_solution: bool = False)[source]
discrete_optimization.workforce.scheduling.problem.satisfy_detailed(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution)[source]
discrete_optimization.workforce.scheduling.problem.satisfy_detailed_all_done(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution)[source]
discrete_optimization.workforce.scheduling.problem.satisfy_detailed_available_team(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution) → list[tuple[str, Hashable, Hashable, int, int]][source]
discrete_optimization.workforce.scheduling.problem.satisfy_detailed_precedence(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution) → list[tuple[str, Hashable, Hashable, int]][source]
discrete_optimization.workforce.scheduling.problem.satisfy_detailed_same_allocation(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution) → list[tuple[str, set[Hashable], set[int]]][source]
discrete_optimization.workforce.scheduling.problem.satisfy_overlap_teams(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, **kwargs) → bool[source]
discrete_optimization.workforce.scheduling.problem.satisfy_overlap_teams_detailed(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution) → list[tuple[str, int, int, int]][source]
discrete_optimization.workforce.scheduling.problem.satisfy_precedence(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, partial_solution: bool = False) → bool[source]

Partial solution = True means we ignore variable set to None when we check the constraint.

discrete_optimization.workforce.scheduling.problem.satisfy_same_allocation(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, partial_solution: bool = False) → bool[source]

Partial solution = True means we ignore variable set to None when we check the constraint. This one is a bit tricky to write when we allow partial solution.

discrete_optimization.workforce.scheduling.problem.satisfy_time_window(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, partial_solution: bool = False)[source]
discrete_optimization.workforce.scheduling.problem.satisfy_time_window_detailed(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution)[source]
discrete_optimization.workforce.scheduling.problem.transform_alloc_solution_to_rcpsp_solution(alloc_solution: AllocSchedulingSolution, rcpsp_problem: RcpspProblem, ac_mode_to_team: dict[tuple[Hashable, int], Hashable], alloc_scheduling_problem: AllocSchedulingProblem) → RcpspSolution[source]
discrete_optimization.workforce.scheduling.problem.transform_rcpsp_solution_to_alloc_solution(rcpsp_solution: RcpspSolution, rcpsp_problem: RcpspProblem, ac_mode_to_team: dict[tuple[Hashable, int], Hashable], alloc_scheduling_problem: AllocSchedulingProblem) → AllocSchedulingSolution[source]
discrete_optimization.workforce.scheduling.problem.transform_to_monomode_rcpsp(problem: AllocSchedulingProblem, build_calendar: bool = True, add_additional_constraint: bool = True) → tuple[RcpspProblem, dict[tuple[Hashable, int], Hashable]][source]
discrete_optimization.workforce.scheduling.problem.transform_to_multimode_rcpsp(problem: AllocSchedulingProblem, build_calendar: bool = True, add_window_time_constraint: bool = True, add_additional_constraint: bool = True) → tuple[RcpspProblem, dict[tuple[Hashable, int], Hashable]][source]

discrete_optimization.workforce.scheduling.utils module

discrete_optimization.workforce.scheduling.utils.alloc_solution_to_alloc_sched_solution(problem: AllocSchedulingProblem, solution: TeamAllocationSolution)[source]
discrete_optimization.workforce.scheduling.utils.binary_calendar(list_available: list[tuple[int, int]], horizon: int)[source]
discrete_optimization.workforce.scheduling.utils.build_allocation_problem_from_scheduling(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution = None, problem_alloc: TeamAllocationProblem = None, multiobjective: bool = True) → TeamAllocationProblem[source]
discrete_optimization.workforce.scheduling.utils.build_scheduling_problem_from_allocation(problem: TeamAllocationProblem, horizon_start_shift: int = 0) → AllocSchedulingProblem[source]
discrete_optimization.workforce.scheduling.utils.compute_available_teams_per_activities_alloc_problem(problem: TeamAllocationProblem, starts: ndarray, ends: ndarray, calendars_team: dict[Hashable, ndarray])[source]
discrete_optimization.workforce.scheduling.utils.compute_changes_between_solution(solution_a: AllocSchedulingSolution, solution_b: AllocSchedulingSolution, problem_a: AllocSchedulingProblem = None, problem_b: AllocSchedulingProblem = None)[source]
discrete_optimization.workforce.scheduling.utils.compute_changes_between_solution_same_pb(solution_a: AllocSchedulingSolution, solution_b: AllocSchedulingSolution, problem: AllocSchedulingProblem = None)[source]
discrete_optimization.workforce.scheduling.utils.compute_equivalent_teams_scheduling_problem(scheduling_problem: AllocSchedulingProblem) → list[list[int]][source]

Return a list of disjoint set of teams index, that can be considered as indistinguishable from a solution point of view. Example : in the pure coloring problem all the colors/team are equivalent In the team allocation problem, due to restricted compatible teams to task, the equivalent class are different Adaptation from the notebook/test_models.ipynb

discrete_optimization.workforce.scheduling.utils.compute_precedence_graph(problem: AllocSchedulingProblem) → DiGraph[source]
discrete_optimization.workforce.scheduling.utils.estimate_nb_resource_needed(problem: AllocSchedulingProblem)[source]
discrete_optimization.workforce.scheduling.utils.export_scheduling_problem_json(problem: AllocSchedulingProblem) → dict[source]
discrete_optimization.workforce.scheduling.utils.get_availability_slots(calendar_matrix: ndarray)[source]
discrete_optimization.workforce.scheduling.utils.get_working_time_teams(problem: AllocSchedulingProblem) → dict[source]
discrete_optimization.workforce.scheduling.utils.natural_explanation_unsat(detailed_output: list[tuple | dict], problem: AllocSchedulingProblem) → list[str][source]
discrete_optimization.workforce.scheduling.utils.natural_explanation_unsat_from_sol(solution: AllocSchedulingSolution) → list[str][source]
discrete_optimization.workforce.scheduling.utils.overlap_interval(interval_1: tuple[int, int], interval_2: tuple[int, int])[source]
discrete_optimization.workforce.scheduling.utils.plot_schedule_comparison(base_solution: AllocSchedulingSolution, updated_solution: AllocSchedulingSolution, problem: AllocSchedulingProblem)[source]

Nice visu to compare 2 schedules.

discrete_optimization.workforce.scheduling.utils.plotly_schedule_comparison(base_solution: AllocSchedulingSolution, updated_solution: AllocSchedulingSolution, problem: AllocSchedulingProblem, index_team_to_other_index: dict[int, int] = None, display: bool = False, additional_info: dict[Hashable, dict[str, Any]] = None, use_color_scale: bool = True, use_color_map_per_task: bool = False, color_map_per_task: dict[int, Any] = None, opacity_map_per_task: dict[int, float] = None, show_all_changes: bool = True, show_change: dict[int, bool] = None, plot_team_breaks: bool = False, plot_xticks: bool = True, plot_text: bool = True, title='Scheduling Comparison (Base vs Updated)')[source]

Nice visu to compare 2 schedules.

discrete_optimization.workforce.scheduling.utils.template_violated_constraint(satisfy_detailed_output: tuple | dict, problem: AllocSchedulingProblem)[source]

Module contents