discrete_optimization.workforce.scheduling package

Subpackages

• discrete_optimization.workforce.scheduling.solvers package
  • Submodules
  • discrete_optimization.workforce.scheduling.solvers.alloc_scheduling_lb module
    • ApproximateBoundAllocScheduling
      • ApproximateBoundAllocScheduling.get_lb_nb_teams()
    • BaseAllocSchedulingLowerBoundProvider
      • BaseAllocSchedulingLowerBoundProvider.get_lb_nb_teams()
      • BaseAllocSchedulingLowerBoundProvider.status
    • BoundResourceViaRelaxedProblem
      • BoundResourceViaRelaxedProblem.get_lb_nb_teams()
      • BoundResourceViaRelaxedProblem.hyperparameters
    • LBoundAllocScheduling
      • LBoundAllocScheduling.get_lb_nb_teams()
  • discrete_optimization.workforce.scheduling.solvers.cpsat module
    • AdditionalCPConstraints
    • CPSatAllocSchedulingSolver
      • CPSatAllocSchedulingSolver.add_lexico_constraint()
      • CPSatAllocSchedulingSolver.add_objective_functions_on_cumul()
      • CPSatAllocSchedulingSolver.create_delta_objectives()
      • CPSatAllocSchedulingSolver.create_makespan_obj()
      • CPSatAllocSchedulingSolver.create_used_variable()
      • CPSatAllocSchedulingSolver.get_lexico_objective_value()
      • CPSatAllocSchedulingSolver.get_lexico_objectives_available()
      • CPSatAllocSchedulingSolver.hyperparameters
      • CPSatAllocSchedulingSolver.implements_lexico_api()
      • CPSatAllocSchedulingSolver.init_model()
      • CPSatAllocSchedulingSolver.problem
      • CPSatAllocSchedulingSolver.retrieve_solution()
      • CPSatAllocSchedulingSolver.set_additional_constraints()
      • CPSatAllocSchedulingSolver.set_lexico_objective()
      • CPSatAllocSchedulingSolver.set_model_obj_aggregated()
      • CPSatAllocSchedulingSolver.set_nb_teams_constraints()
      • CPSatAllocSchedulingSolver.set_team_used_constraint()
      • CPSatAllocSchedulingSolver.set_warm_start()
      • CPSatAllocSchedulingSolver.variables
  • discrete_optimization.workforce.scheduling.solvers.cpsat_relaxed module
    • CPSatAllocSchedulingSolverCumulative
      • CPSatAllocSchedulingSolverCumulative.add_lexico_constraint()
      • CPSatAllocSchedulingSolverCumulative.create_makespan_obj()
      • CPSatAllocSchedulingSolverCumulative.define_objectives()
      • CPSatAllocSchedulingSolverCumulative.get_lexico_objective_value()
      • CPSatAllocSchedulingSolverCumulative.get_lexico_objectives_available()
      • CPSatAllocSchedulingSolverCumulative.hyperparameters
      • CPSatAllocSchedulingSolverCumulative.implements_lexico_api()
      • CPSatAllocSchedulingSolverCumulative.init_main_vars()
      • CPSatAllocSchedulingSolverCumulative.init_model()
      • CPSatAllocSchedulingSolverCumulative.init_multimode_data()
      • CPSatAllocSchedulingSolverCumulative.not_implemented_objectives
      • CPSatAllocSchedulingSolverCumulative.problem
      • CPSatAllocSchedulingSolverCumulative.retrieve_solution()
      • CPSatAllocSchedulingSolverCumulative.set_lexico_objective()
      • CPSatAllocSchedulingSolverCumulative.set_model_obj_aggregated()
      • CPSatAllocSchedulingSolverCumulative.set_precedence_constraints()
      • CPSatAllocSchedulingSolverCumulative.set_resource_pool_constraints()
      • CPSatAllocSchedulingSolverCumulative.set_same_allocation_constraints()
      • CPSatAllocSchedulingSolverCumulative.set_warm_start()
      • CPSatAllocSchedulingSolverCumulative.solve()
      • CPSatAllocSchedulingSolverCumulative.variables
  • Module contents
    • ObjectivesEnum
      • ObjectivesEnum.DELTA_TO_EXISTING_SOLUTION
      • ObjectivesEnum.DISPERSION
      • ObjectivesEnum.MAKESPAN
      • ObjectivesEnum.MIN_WORKLOAD
      • ObjectivesEnum.NB_DONE_AC
      • ObjectivesEnum.NB_TEAMS
    • SolverAllocScheduling
      • SolverAllocScheduling.problem

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]

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

discrete_optimization.workforce.scheduling.parser.update_calendars_with_disruptions(list_drop_resource, teams: list[Hashable], calendar_team: dict[Hashable, list[tuple[int, int]]])

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)

Bases: Problem

compatible_teams_all_activity() → dict[Hashable, set[Hashable]]

compatible_teams_index_all_activity() → dict[int, set[int]]

compute_predecessors()

compute_unavailability_calendar(team: Hashable) → list[tuple[int, int]]

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

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

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_attribute_register() → EncodingRegister

Returns how the Solution should be encoded.

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

get_lb_end_window(task: Hashable) → int

get_lb_start_window(task: Hashable) → int

get_objective_register() → ObjectiveRegister

Returns the objective definition.

Returns (ObjectiveRegister): object defining the objective criteria.

get_solution_type() → type[Solution]

Returns the class implementation of a Solution.

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

get_ub_end_window(task: Hashable) → int

get_ub_start_window(task: Hashable) → int

get_unavailable_teams_per_activity() → dict[Hashable, set[Hashable]]

satisfy(variable: Solution) → bool

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)

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

Override the available team attribute and update the rcpsp accordingly

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

Bases: Solution

change_problem(new_problem: Problem) → None

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

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.

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

Bases: object

discrete_optimization.workforce.scheduling.problem.build_calendar_array_from_availability_slot(availability_slots: list[tuple[int, int]], horizon: int, value: int = 1)

discrete_optimization.workforce.scheduling.problem.build_pair_mode_constraint(problem: AllocSchedulingProblem, rcpsp: RcpspProblem, use_score: bool = False)

discrete_optimization.workforce.scheduling.problem.compute_stats_per_team(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution) → dict[int, float]

discrete_optimization.workforce.scheduling.problem.correct_schedule_avoid_overlap(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, init_min_starting_date_lb: bool = False)

discrete_optimization.workforce.scheduling.problem.evaluate_solution(solution: AllocSchedulingSolution, problem: AllocSchedulingProblem) → dict[str, float]

discrete_optimization.workforce.scheduling.problem.export_scheduling_problem_json(problem: AllocSchedulingProblem) → dict

discrete_optimization.workforce.scheduling.problem.full_satisfy(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, partial_solution: bool = False) → bool

discrete_optimization.workforce.scheduling.problem.interval_inside(interval1: tuple[float, float], interval_container: tuple[float, float])

discrete_optimization.workforce.scheduling.problem.intervals_do_not_overlap(interval1: tuple[float, float], interval2: tuple[float, float])

discrete_optimization.workforce.scheduling.problem.realign_calendars(calendars_dict: dict[Hashable, list[tuple[int, int]]])

discrete_optimization.workforce.scheduling.problem.satisfy_all_done(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, partial_solution: bool = False)

discrete_optimization.workforce.scheduling.problem.satisfy_available_team(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, partial_solution: bool = False) → bool

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

discrete_optimization.workforce.scheduling.problem.satisfy_calendars(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, partial_solution: bool = False)

discrete_optimization.workforce.scheduling.problem.satisfy_detailed(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution)

discrete_optimization.workforce.scheduling.problem.satisfy_detailed_all_done(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution)

discrete_optimization.workforce.scheduling.problem.satisfy_detailed_available_team(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution) → list[tuple[str, Hashable, Hashable, int, int]]

discrete_optimization.workforce.scheduling.problem.satisfy_detailed_precedence(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution) → list[tuple[str, Hashable, Hashable, int]]

discrete_optimization.workforce.scheduling.problem.satisfy_detailed_same_allocation(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution) → list[tuple[str, set[Hashable]], set[int]]

discrete_optimization.workforce.scheduling.problem.satisfy_overlap_teams(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, **kwargs) → bool

discrete_optimization.workforce.scheduling.problem.satisfy_overlap_teams_detailed(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution) → list[tuple[str, int, int, int]]

discrete_optimization.workforce.scheduling.problem.satisfy_precedence(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution, partial_solution: bool = False) → bool

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

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)

discrete_optimization.workforce.scheduling.problem.satisfy_time_window_detailed(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution)

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

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

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

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

discrete_optimization.workforce.scheduling.utils module

discrete_optimization.workforce.scheduling.utils.alloc_solution_to_alloc_sched_solution(problem: AllocSchedulingProblem, solution: TeamAllocationSolution)

discrete_optimization.workforce.scheduling.utils.binary_calendar(list_available: list[tuple[int, int]], horizon: int)

discrete_optimization.workforce.scheduling.utils.build_allocation_problem_from_scheduling(problem: AllocSchedulingProblem, solution: AllocSchedulingSolution | None = None, problem_alloc: TeamAllocationProblem | None = None, multiobjective: bool = True) → TeamAllocationProblem

discrete_optimization.workforce.scheduling.utils.build_scheduling_problem_from_allocation(problem: TeamAllocationProblem, horizon_start_shift: int = 0) → AllocSchedulingProblem

discrete_optimization.workforce.scheduling.utils.compute_available_teams_per_activities_alloc_problem(problem: TeamAllocationProblem, starts: ndarray, ends: ndarray, calendars_team: dict[Hashable, ndarray])

discrete_optimization.workforce.scheduling.utils.compute_changes_between_solution(solution_a: AllocSchedulingSolution, solution_b: AllocSchedulingSolution, problem_a: AllocSchedulingProblem | None = None, problem_b: AllocSchedulingProblem | None = None)

discrete_optimization.workforce.scheduling.utils.compute_changes_between_solution_same_pb(solution_a: AllocSchedulingSolution, solution_b: AllocSchedulingSolution, problem: AllocSchedulingProblem | None = None)

discrete_optimization.workforce.scheduling.utils.compute_equivalent_teams_scheduling_problem(scheduling_problem: AllocSchedulingProblem) → list[list[int]]

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

discrete_optimization.workforce.scheduling.utils.estimate_nb_resource_needed(problem: AllocSchedulingProblem)

discrete_optimization.workforce.scheduling.utils.export_scheduling_problem_json(problem: AllocSchedulingProblem) → dict

discrete_optimization.workforce.scheduling.utils.get_availability_slots(calendar_matrix: ndarray)

discrete_optimization.workforce.scheduling.utils.get_working_time_teams(problem: AllocSchedulingProblem) → dict

discrete_optimization.workforce.scheduling.utils.natural_explanation_unsat(detailed_output: list[tuple | dict], problem: AllocSchedulingProblem) → list[str]

discrete_optimization.workforce.scheduling.utils.natural_explanation_unsat_from_sol(solution: AllocSchedulingSolution) → list[str]

discrete_optimization.workforce.scheduling.utils.overlap_interval(interval_1: tuple[int, int], interval_2: tuple[int, int])

discrete_optimization.workforce.scheduling.utils.plot_schedule_comparison(base_solution: AllocSchedulingSolution, updated_solution: AllocSchedulingSolution, problem: AllocSchedulingProblem)

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 = None, display: bool = False, additional_info: dict[Hashable, dict[str, Any]] | None = None, use_color_scale: bool = True, use_color_map_per_task: bool = False, color_map_per_task: dict[int, Any] | None = None, opacity_map_per_task: dict[int, float] | None = None, show_all_changes: bool = True, show_change: dict[int, bool] | None = None, plot_team_breaks: bool = False, plot_xticks: bool = True, plot_text: bool = True, title='Scheduling Comparison (Base vs Updated)')

Nice visu to compare 2 schedules.

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

Module contents