discrete_optimization.flex_scheduling.solvers package

Submodules

discrete_optimization.flex_scheduling.solvers.cpsat module

class discrete_optimization.flex_scheduling.solvers.cpsat.ConstraintIncluding(include_non_released_resource: bool = True, include_group_variables: bool = True, include_constraint_precedence_on_groups: bool = True, include_constraints_on_groups: bool = True, include_generalized_time_constraints: bool = True, include_variable_resource: bool = True)

Bases: object

include_constraint_precedence_on_groups: bool = True

include_constraints_on_groups: bool = True

include_generalized_time_constraints: bool = True

include_group_variables: bool = True

include_non_released_resource: bool = True

include_variable_resource: bool = True

class discrete_optimization.flex_scheduling.solvers.cpsat.CpSatFlexSolver(problem: FlexProblem, **kwargs: Any)

Bases: PrecedenceSchedulingCpSatSolver[Hashable], CumulativeResourceSchedulingCpSatSolver[Hashable, Hashable, None], NonRenewableCpSatSolver[Hashable, Hashable], WarmstartMixin

add_lexico_constraint(obj: str, value: float) → Iterable[Any]

Add a constraint on a computed sub-objective

Parameters:

• obj – a string representing the desired objective. Should be one of self.get_lexico_objectives_available().

• value – the limiting value. If the optimization direction is maximizing, this is a lower bound, else this is an upper bound.

Returns:

the created constraints.

constraint_cumulative(constraint_including: ConstraintIncluding)

constraint_cumulative_resource(resource: ResourceData, variable_max_capacity: bool = False, constraint_including: ConstraintIncluding = None)

constraint_cumulative_resource_depr(resource: ResourceData, variable_max_capacity: bool = False)

Simple and tested version

constraint_duration_of_task(task_index: int, mode: int, duration_per_interval: Dict[int, List[Tuple[int, int]]])

constraint_duration_of_task_element(task_index: int, mode: int)

constraint_duration_of_tasks()

Tricky constraint : should take into account the partial preemption possibility, which makes duration variable based on calendars

constraint_generalized_time_constraint()

constraint_group_non_release_resource()

constraint_group_non_release_resource_reservoir()

constraint_non_overlap_group()

constraint_non_renewable_resource(resource: ResourceData, variable_max_capacity: bool = False)

constraint_on_groups_of_task()

constraint_precedence()

Basic precedence constraint

constraint_precedence_on_groups()

Basic precedence constraint between group of tasks

constraint_reservoir(constraint_including: ConstraintIncluding = None)

constraint_sequencing_on_disjunctive_resource()

create_earliness_objective(obj_earliness: ObjectiveParamEarliness)

create_makespan()

create_objectives(constraint_including: ConstraintIncluding = None)

create_resource_objective(obj_params_resource: ObjectiveParamResource)

create_tardiness_objective(obj_tardiness: ObjectiveParamTardiness)

duration_encoding: DurationEncodingEnum

durs: Dict[Tuple[int, int], Tuple[List[int], Dict[int, List[Tuple[int, int]]]]]

from_solution_to_hint(solution: ScheduleSolution) → list[tuple[IntVar | int | int8 | uint8 | int32 | uint32 | int64 | uint64, int]]

from_solution_to_hint_earliness(solution: ScheduleSolution)

from_solution_to_hint_non_released_delta(solution: ScheduleSolution)

get_lexico_objective_value(obj: str, res: ResultStorage) → float

Get best internal model objective value found by last call to solve().

The default implementation consists in using the fit of the last solution in result_storage. This assumes: - that the last solution is the best one for the objective considered - that no aggregation was performed but rather that the fitness is a TupleFitness

with values in the same order as self.problem.get_objective_names().

Parameters:

• obj – a string representing the desired objective. Should be one of self.get_lexico_objectives_available().

• res – result storage returned by last call to solve().

Returns:

get_lexico_objectives_available() → list[str]

List objectives available for lexico optimization

It corresponds to the labels accepted for obj argument for - set_lexico_objective() - add_lexico_constraint() - get_lexico_objective_value()

Default to self.problem.get_objective_names().

Returns:

get_objr_expr(obj: str | tuple)

get_task_mode_interval(task: Hashable, mode: int) → IntervalVar

Get the interval variable corresponding to given task and mode.

Needs to be created if avoid_interval_optional is False. Not necessary else.

get_task_mode_is_present_variable(task: Hashable, mode: int) → LinearExpr | IntVar | int | int8 | uint8 | int32 | uint32 | int64 | uint64

Retrieve the 0-1 variable/expression telling if the mode is used for the task.

Parameters:

• task

• mode

Returns:

get_task_start_or_end_variable(task: Hashable, start_or_end: StartOrEnd) → LinearExpr | IntVar | int | int8 | uint8 | int32 | uint32 | int64 | uint64

Retrieve the variable storing the start or end time of given task.

Parameters:

• task

• start_or_end

Returns:

get_task_unary_resource_interval(task: Hashable, unary_resource: UnaryResource) → IntervalVar

get_task_unary_resource_is_present_variable(task: Hashable, unary_resource: UnaryResource) → LinearExpr | IntVar | int | int8 | uint8 | int32 | uint32 | int64 | uint64

hyperparameters: list[Hyperparameter] = [EnumHyperparameter(name='duration_encoding', default=<DurationEncodingEnum.INDICATOR: 0>, depends_on=None, name_in_kwargs='duration_encoding')]

Hyperparameters available for this solver.

These hyperparameters are to be feed to **kwargs found in

• __init__()

• init_model() (when available)

• solve()

implements_lexico_api() → bool

Tell whether this solver is implementing the api for lexicographic optimization.

Should return True only if

• set_lexico_objective()

• add_lexico_constraint()

• get_lexico_objective_value()

have been really implemented, i.e. - calling set_lexico_objective() and add_lexico_constraint()

should actually change the next call to solve(),

• get_lexico_objective_value() should correspond to the internal model objective

init_and_constraint_wip_variables()

init_group_variables()

init_intervals_of_non_released_resource(constraint_including: ConstraintIncluding = None)

init_main_variables()

init_model(constraint_including: ConstraintIncluding | None = None, **args: Any) → None

Init cp model and reset stored variables if any.

init_optional_interval_variables()

init_resource_variables()

problem: FlexProblem

retrieve_solution(cpsolvercb: CpSolverSolutionCallback) → ScheduleSolution

Construct a do solution from the cpsat solver internal solution.

It will be called each time the cpsat solver find a new solution. At that point, value of internal variables are accessible via cpsolvercb.Value(VARIABLE_NAME).

Parameters:

cpsolvercb – the ortools callback called when the cpsat solver finds a new solution.

Returns:

the intermediate solution, at do format.

set_lexico_objective(obj: str) → None

Update internal model objective.

Parameters:

obj – a string representing the desired objective. Should be one of self.get_lexico_objectives_available().

Returns:

set_warm_start(solution: Solution) → None

Make the solver warm start from the given solution.

set_warm_start_from_sol(solution: Solution) → None

class discrete_optimization.flex_scheduling.solvers.cpsat.DurationEncodingEnum(*values)

Bases: Enum

ELEMENT = 1

INDICATOR = 0

discrete_optimization.flex_scheduling.solvers.cpsat.build_multiple_cumulative_constraints_inputs(problem: FlexProblem, resource: ResourceData)

This is a utility function to define the set of interval variable to include in separated cumulative constraints.

this handle corner case where : non-zero resource availability happens in some ressource, and this non-zero level is lower than the resource need of the task. -When this happen, it shouldn’t prevent the task to overlap with this period of time -> it just means that the task has a longer duration, as computed in compute_duration_tasks_function_time function. -With previous formulation of cumulative constraint, this corner case is not well taken into account, We put in a cumulative constraint all task consuming some resource and all interval corresponding to (partial) resource unavaibility, and this was leading to unsat problem

discrete_optimization.flex_scheduling.solvers.cpsat.compute_duration_function_time_cluster(orig_duration: int, resource_calendar: ndarray, cumulative_resource_calendar: ndarray)

discrete_optimization.flex_scheduling.solvers.cpsat.compute_duration_tasks_function_time(problem: FlexProblem, method=None)

discrete_optimization.flex_scheduling.solvers.cpsat.post_cumulative_constraints(problem: FlexProblem, resource: ResourceData, solver: CpSatFlexSolver, variable_max_capacity: bool, include_intervals_non_release: bool = True)

discrete_optimization.flex_scheduling.solvers.cpsat_preempt module

class discrete_optimization.flex_scheduling.solvers.cpsat_preempt.CPSatFlexSPPreempt(problem: FlexProblem, **kwargs: Any)

Bases: CpSatFlexSolver

compute_possible_start_end_duration_values(index_task: int)

constraint_convention_preemption()

constraint_cumulative()

constraint_cumulative_resource(resource: ResourceData, variable_max_capacity: bool = False)

constraint_group_non_release_resource()

constraint_sum_duration()

create_objectives()

from_solution_to_hint(solution: ScheduleSolutionPreemptive) → list[tuple[IntVar | int | int8 | uint8 | int32 | uint32 | int64 | uint64, int]]

from_solution_to_hint_earliness(solution: ScheduleSolutionPreemptive)

from_solution_to_hint_non_released_delta(solution: ScheduleSolutionPreemptive)

get_skill_variable(task: Task, unary_resource: UnaryResource, skill: Skill) → LinearExpr | IntVar | int | int8 | uint8 | int32 | uint32 | int64 | uint64

init_main_variables()

init_main_variables_preempt(nb_max_preemption: int = None)

init_model(**args: Any) → None

Init cp model and reset stored variables if any.

init_modes_variables()

init_span_task_variables()

init_variable_preempt_mode()

retrieve_solution(cpsolvercb: CpSolverSolutionCallback) → ScheduleSolution

Construct a do solution from the cpsat solver internal solution.

It will be called each time the cpsat solver find a new solution. At that point, value of internal variables are accessible via cpsolvercb.Value(VARIABLE_NAME).

Parameters:

cpsolvercb – the ortools callback called when the cpsat solver finds a new solution.

Returns:

the intermediate solution, at do format.

discrete_optimization.flex_scheduling.solvers.cpsat_preempt.resource_consumption(flex_problem: FlexProblem, solution: ScheduleSolution)

discrete_optimization.flex_scheduling.solvers.heuristic_solver module

class discrete_optimization.flex_scheduling.solvers.heuristic_solver.HeuristicSolverFlexProblem(problem: FlexProblem, params_objective_function: ParamsObjectiveFunction | None = None, **kwargs: Any)

Bases: SolverDO, WarmstartMixin

hyperparameters: list[Hyperparameter] = []

Hyperparameters available for this solver.

These hyperparameters are to be feed to **kwargs found in

• __init__()

• init_model() (when available)

• solve()

modify_calendars()

put_back_calendar()

set_warm_start(solution: Solution) → None

Make the solver warm start from the given solution.

solve(callbacks: list[Callback] | None = None, params_cp_first_solve: ParametersCp = None, params_cp_second_solve: ParametersCp = None, time_limit_per_objectives_first_solve: int = 150, time_limit_per_objectives_second_solve: int = 150, objectives_first_solve: list[str] = None, objectives_second_solve: list[str] = None, **kwargs: Any) → ResultStorage

Generic solving function.

Parameters:

• callbacks – list of callbacks used to hook into the various stage of the solve

• **kwargs – any argument specific to the solver

Solvers deriving from SolverDo should use callbacks methods .on_step_end(), … during solve(). But some solvers are not yet updated and are just ignoring it.

Returns (ResultStorage): a result object containing potentially a pool of solutions to a discrete-optimization problem

class discrete_optimization.flex_scheduling.solvers.heuristic_solver.WSCallback(objectives: list[str])

Bases: Callback

on_step_end(step: int, res: ResultStorage, solver: SolverDO)

Called at the end of an optimization step.

Parameters:

• step – index of step

• res – current result storage

• solver – solvers using the callback

Returns:

If True, the optimization process is stopped, else it goes on.

discrete_optimization.flex_scheduling.solvers.optal module

class discrete_optimization.flex_scheduling.solvers.optal.ConstraintIncludingParams(include_calendar_and_duration: bool = True, include_non_released_resource: bool = True, include_group_variables: bool = True, include_constraint_precedence_on_groups: bool = True, include_constraints_on_groups: bool = True, include_generalized_time_constraints: bool = True, include_variable_resource: bool = True, add_precedence_non_release_variables: bool = True, include_cumulative_constraint: bool = True, include_reservoir_constraint_non_release: bool = False, max_length_non_release: int | None = None, synchro_instead_of_non_release: bool = False)

Bases: object

add_precedence_non_release_variables: bool = True

include_calendar_and_duration: bool = True

include_constraint_precedence_on_groups: bool = True

include_constraints_on_groups: bool = True

include_cumulative_constraint: bool = True

include_generalized_time_constraints: bool = True

include_group_variables: bool = True

include_non_released_resource: bool = True

include_reservoir_constraint_non_release: bool = False

include_variable_resource: bool = True

max_length_non_release: int | None = None

synchro_instead_of_non_release: bool = False

class discrete_optimization.flex_scheduling.solvers.optal.OptalFlexProblemSolver(problem: FlexProblem, params_objective_function: ParamsObjectiveFunction | None = None, **args)

Bases: OptalCpSolver

add_duration_constraint_element()

add_duration_constraint_integral()

add_duration_constraint_no_calendar()

add_lexico_constraint(obj: str, value: float) → Iterable[Any]

Add a constraint on a computed sub-objective

Parameters:

• obj – a string representing the desired objective. Should be one of self.get_lexico_objectives_available().

• value – the limiting value. If the optimization direction is maximizing, this is a lower bound, else this is an upper bound.

Returns:

the created constraints.

alternative_modes()

constraint_cumulative(params: ConstraintIncludingParams)

constraint_cumulative_resource(resource: ResourceData, variable_max_capacity: bool = False)

constraint_group_non_release_resource()

constraint_non_overlap_group()

constraint_non_renewable_resource(resource: ResourceData, variable_max_capacity: bool = False)

constraint_precedence()

Basic precedence constraint

constraint_precedence_on_groups()

Basic precedence constraint between group of tasks

constraint_reservoir_non_release_res(resource: ResourceData)

create_base_intervals()

create_calendar_step_function()

create_earliness_objective(obj_earliness: ObjectiveParamEarliness)

create_makespan()

create_objectives(params: ConstraintIncludingParams)

create_opt_intervals()

create_resource_objective(obj_params_resource: ObjectiveParamResource)

create_resource_release_cost()

create_synchro_of_non_released_resource(params: ConstraintIncludingParams)

create_tardiness_objective(obj_tardiness: ObjectiveParamTardiness)

current_objective: str

get_lexico_objective_value(obj: str, res: ResultStorage) → float

Get best internal model objective value found by last call to solve().

The default implementation consists in using the fit of the last solution in result_storage. This assumes: - that the last solution is the best one for the objective considered - that no aggregation was performed but rather that the fitness is a TupleFitness

with values in the same order as self.problem.get_objective_names().

Parameters:

• obj – a string representing the desired objective. Should be one of self.get_lexico_objectives_available().

• res – result storage returned by last call to solve().

Returns:

get_lexico_objectives_available() → list[str]

List objectives available for lexico optimization

It corresponds to the labels accepted for obj argument for - set_lexico_objective() - add_lexico_constraint() - get_lexico_objective_value()

Default to self.problem.get_objective_names().

Returns:

get_objr_expr(obj: str | tuple)

implements_lexico_api() → bool

Tell whether this solver is implementing the api for lexicographic optimization.

Should return True only if

• set_lexico_objective()

• add_lexico_constraint()

• get_lexico_objective_value()

have been really implemented, i.e. - calling set_lexico_objective() and add_lexico_constraint()

should actually change the next call to solve(),

• get_lexico_objective_value() should correspond to the internal model objective

init_group_variables()

init_intervals_of_non_released_resource(params: ConstraintIncludingParams)

init_model(params: ConstraintIncludingParams = None, **args: Any) → None

Instantiate a CP model instance

Afterwards, self.instance should not be None anymore.

init_resource_variables()

problem: FlexProblem

retrieve_sol_from_solver_solution(solution: cp.Solution)

retrieve_solution(result: cp.SolveResult) → ScheduleSolution

Return a d-o solution from the variables computed by minizinc.

Parameters:

result – output of the cp.solve

Returns:

set_lexico_objective(obj: str) → None

Update internal model objective.

Parameters:

obj – a string representing the desired objective. Should be one of self.get_lexico_objectives_available().

Returns:

set_warm_start(solution: ScheduleSolution) → None

Creates an OptalCP Solution object from a discrete-optimization ScheduleSolution and registers it as a warm start for the solver.

set_warm_start_from_previous_run()

Set warm start from previous run of the solver.

discrete_optimization.flex_scheduling.solvers.optal.compute_duration_tasks_function_time_and_resource_calendars(problem: FlexProblem) → tuple[Any, dict[tuple, ndarray], dict[tuple[int, int], tuple]]

discrete_optimization.flex_scheduling.solvers.optal.post_cumulative_constraints(problem: FlexProblem, resource: ResourceData, solver: OptalFlexProblemSolver, variable_max_capacity: bool, include_intervals_non_release: bool = True)

discrete_optimization.flex_scheduling.solvers.sequential_solver module

class discrete_optimization.flex_scheduling.solvers.sequential_solver.SequentialFlexSolver(problem: FlexProblem, params_objective_function: ParamsObjectiveFunction | None = None, **kwargs)

Bases: SolverDO

solve(nb_batches: int = 5, time_limit_per_batch: float = 60.0, **kwargs) → ResultStorage

Solves the problem iteratively by growing the problem scope batch by batch.

Ensures that the final solution is re-ordered to match the original problem’s task order.

Module contents