discrete_optimization.generic_tasks_tools package

Subpackages

Submodules

discrete_optimization.generic_tasks_tools.allocation module

class discrete_optimization.generic_tasks_tools.allocation.AllocationCpSolver(problem: Problem, params_objective_function: ParamsObjectiveFunction | None = None, **kwargs: Any)[source]

Bases: TasksCpSolver[Task], Generic[Task, UnaryResource]

Base class for solver managing constraints on allocation.

abstractmethod add_constraint_on_nb_allocation_changes(ref: AllocationSolution[Task, UnaryResource], nb_changes: int) → list[Any][source]

Add contraint on maximal number of allocation changes from the given reference.

Parameters:

  • ref

  • nb_changes – maximal number of changes

Returns:

resulting constraints

abstractmethod add_constraint_on_task_unary_resource_allocation(task: Task, unary_resource: UnaryResource, used: bool) → list[Any][source]

Add constraint on allocation of given unary resource for the given task

Parameters:

  • task

  • unary_resource

  • used – if True, we enforce the allocation of unary_resource to task, else we prevent it

Returns:

resulting constraints

abstractmethod add_constraint_on_total_nb_usages(sign: SignEnum, target: int) → list[Any][source]
abstractmethod add_constraint_on_unary_resource_nb_usages(unary_resource: UnaryResource, sign: SignEnum, target: int) → list[Any][source]
add_constraint_same_allocation_as_ref(ref: AllocationSolution[Task, UnaryResource], tasks: Iterable[Task] | None = None, unary_resources: Iterable[UnaryResource] | None = None) → list[Any][source]

Add constraint to keep same allocation as the reference for the given tasks and unary resources subsets.

Parameters:

  • ref

  • tasks

  • unary_resources

Returns:

resulting constraints

get_default_tasks_n_unary_resources(tasks: Iterable[Task] | None = None, unary_resources: Iterable[UnaryResource] | None = None) → tuple[Iterable[Task], Iterable[UnaryResource]][source]
abstractmethod get_nb_tasks_done_variable() → Any[source]

Construct and get the variable tracking number of tasks with at least a resource allocated.

Returns:

objective variable to minimize

abstractmethod get_nb_unary_resources_used_variable() → Any[source]

Construct and get the variable tracking number of tasks with at least a resource allocated.

Returns:

objective variable to minimize

problem: AllocationProblem[Task, UnaryResource]
class discrete_optimization.generic_tasks_tools.allocation.AllocationProblem[source]

Bases: TasksProblem[Task], Generic[Task, UnaryResource]

Base class for allocation problems.

An allocation problems consist in allocating resources to tasks.

get_index_from_unary_resource(unary_resource: UnaryResource) → int[source]
get_unary_resource_from_index(i: int) → UnaryResource[source]
is_compatible_task_unary_resource(task: Task, unary_resource: UnaryResource) → 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.

abstract property unary_resources_list: list[UnaryResource]

Available unary resources.

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

class discrete_optimization.generic_tasks_tools.allocation.AllocationSolution(problem: Problem)[source]

Bases: TasksSolution[Task], Generic[Task, UnaryResource]

Class inherited by a solution for allocation problems.

check_same_allocation_as_ref(ref: AllocationSolution[Task, UnaryResource], tasks: Iterable[Task] | None = None, unary_resources: Iterable[UnaryResource] | None = None) → bool[source]
compute_nb_allocation_changes(ref: AllocationSolution[Task, UnaryResource], tasks: Iterable[Task] | None = None, unary_resources: Iterable[UnaryResource] | None = None) → int[source]
compute_nb_tasks_done() → int[source]

Compute number of tasks with at least a resource allocated.

compute_nb_unary_resource_usages(tasks: Iterable[Task] | None = None, unary_resources: Iterable[UnaryResource] | None = None)[source]
compute_nb_unary_resources_used() → int[source]

Compute number of unary resources allocated to at least one task.

get_default_tasks_n_unary_resources(tasks: Iterable[Task] | None = None, unary_resources: Iterable[UnaryResource] | None = None) → tuple[Iterable[Task], Iterable[UnaryResource]][source]
abstractmethod is_allocated(task: Task, unary_resource: UnaryResource) → bool[source]

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

Parameters:

  • task

  • unary_resource

Returns:

problem: AllocationProblem[Task, UnaryResource]
discrete_optimization.generic_tasks_tools.allocation.get_default_tasks_n_unary_resources(problem: AllocationProblem, tasks: Iterable[Task] | None = None, unary_resources: Iterable[UnaryResource] | None = None) → tuple[Iterable[Task], Iterable[UnaryResource]][source]

discrete_optimization.generic_tasks_tools.allocation_scheduling module

class discrete_optimization.generic_tasks_tools.allocation_scheduling.AllocationSchedulingProblem[source]

Bases: CumulativeResourceProblem[Task, CumulativeResource | UnaryResource], AllocationProblem[Task, UnaryResource], Generic[Task, UnaryResource, CumulativeResource]

Scheduling problem with unary resource allocation.

This class derives from other mixins to provide utilities that require that mix: - renewable: the unary resources have their own calendar that will be used for constraining allocations - multimode: the tasks have several mode on which the duration depends - cumulative: the tasks consume cumulative resources according to the chosen mode - allocation

Even though this class is generic but encompasses also more specific cases: - singlemode: actually only one mode per task - no cumulative ressources: if resources_list list only unary resources - no calendar: resource capacity can be given as a constant on [0, horizon)

We suppose that all renewable resources are - either cumulative ones - or unary resources

This generic class is to be used to construct generic solvers (e.g. cpsat) that will require no methods implementation to work.

check_renewable_resources_list() → None[source]

Check renewable resources list.

Raises:

AssertionError – if duplicates appear in the list

Returns:

abstract property cumulative_resources_list: list[CumulativeResource]
is_cumulative_resource(resource: CumulativeResource | UnaryResource) → bool[source]

Check if given resource is a cumulative resource.

is_unary_resource(resource: CumulativeResource | UnaryResource) → bool[source]

Check if given resource is a unary resource.

property renewable_resources_list: list[CumulativeResource | UnaryResource]

Renewable resources used by the tasks.

Notes

  • renewable = the resource replenishes as soon as a task using it ends;

  • it can be a mix of unary resources (e.g. employees) and cumulative resources (e.g. tool types).

update_resource_availabilities() → None[source]

Method to call when the resource availabilities have changed.

Default implementation clears the cache on get_resource_max_capacity().

class discrete_optimization.generic_tasks_tools.allocation_scheduling.AllocationSchedulingSolution(problem: Problem)[source]

Bases: CumulativeResourceSolution[Task, CumulativeResource | UnaryResource], AllocationSolution[Task, UnaryResource], Generic[Task, UnaryResource, CumulativeResource]

Solution type associated to AllocationSchedulingProblem.

get_resource_consumption(resource: CumulativeResource | UnaryResource, task: Task) → int[source]
problem: AllocationSchedulingProblem[Task, UnaryResource, CumulativeResource]

discrete_optimization.generic_tasks_tools.base module

class discrete_optimization.generic_tasks_tools.base.TasksCpSolver(problem: Problem, params_objective_function: ParamsObjectiveFunction | None = None, **kwargs: Any)[source]

Bases: CpSolver, Generic[Task]

Base class for cp solver handling tasks problems.

problem: TasksProblem[Task]
class discrete_optimization.generic_tasks_tools.base.TasksProblem[source]

Bases: Problem, Generic[Task]

Base class for scheduling/allocation problems.

abstract property tasks_list: list[Task]

List of all tasks to schedule or allocate to.

class discrete_optimization.generic_tasks_tools.base.TasksSolution(problem: Problem)[source]

Bases: Solution, Generic[Task]

Base class for sheduling/allocation solutions.

problem: TasksProblem[Task]

discrete_optimization.generic_tasks_tools.cumulative_resource module

class discrete_optimization.generic_tasks_tools.cumulative_resource.CumulativeResourceProblem[source]

Bases: RenewableResourceProblem[Task, Resource], MultimodeSchedulingProblem[Task], Generic[Task, Resource]

Scheduling problem with cumulative resources consumed by task.

This derives from problem with renewable resources, some of them are cumulative, some are not (e.g. unary resource if it is moreover an allocation problem). The task consumption of these cumulative resources is supposed to be determined entirely determined by the task mode.

abstractmethod get_resource_consumption(resource: Resource, task: Task, mode: int) → int[source]

Get resource consumption of the task in the given mode

Parameters:

  • resource

  • task

  • mode – not used for single mode problems

Returns:

the consumption for cumulative resources.

Raises:

ValueError – if resource consumption is depending on other variables than mode

abstractmethod is_cumulative_resource(resource: Resource) → bool[source]

Check if given resource is a cumulative resource whose consumption depends only on task mode.

Parameters:

resource

Returns:

class discrete_optimization.generic_tasks_tools.cumulative_resource.CumulativeResourceSolution(problem: Problem)[source]

Bases: RenewableResourceSolution[Task, Resource], MultimodeSchedulingSolution[Task], Generic[Task, Resource]

Solution type associated to CumulativeResourceProblem.

get_resource_consumption(resource: Resource, task: Task) → int[source]

Get resource consumption by given task.

Default implementation works only for cumulative resources whose consumptions depend only on task mode.

Parameters:

  • resource

  • task

Returns:

problem: CumulativeResourceProblem[Task, Resource]

discrete_optimization.generic_tasks_tools.enums module

class discrete_optimization.generic_tasks_tools.enums.StartOrEnd(*values)[source]

Bases: Enum

END = 'end'
START = 'start'

discrete_optimization.generic_tasks_tools.multimode module

class discrete_optimization.generic_tasks_tools.multimode.MultimodeCpSolver(problem: Problem, params_objective_function: ParamsObjectiveFunction | None = None, **kwargs: Any)[source]

Bases: TasksCpSolver[Task]

Class inherited by a solver managing constraints on tasks modes.

abstractmethod add_constraint_on_task_mode(task: Task, mode: int) → list[Any][source]

Add constraint on task mode

The mode of task is fixed to mode.

Parameters:

  • task

  • mode

Returns:

resulting constraints

problem: MultimodeProblem[Task]
class discrete_optimization.generic_tasks_tools.multimode.MultimodeProblem[source]

Bases: TasksProblem[Task]

Class inherited by a solution exposing tasks modes.

abstractmethod get_task_modes(task: Task) → set[int][source]

Retrieve mode found for given task.

Parameters:

task

Returns:

property is_multimode: bool
property max_number_of_mode: int
class discrete_optimization.generic_tasks_tools.multimode.MultimodeSolution(problem: Problem)[source]

Bases: TasksSolution[Task]

Class inherited by a solution exposing tasks modes.

abstractmethod get_mode(task: Task) → int[source]

Retrieve mode found for given task.

Parameters:

task

Returns:

problem: MultimodeProblem[Task]
class discrete_optimization.generic_tasks_tools.multimode.SinglemodeProblem[source]

Bases: MultimodeProblem[Task]

property default_mode

Default single mode.

To be overriden when default value has more sense with another value (ex: in rcpsp, default mode is 1)

get_task_modes(task: Task) → set[int][source]

Retrieve mode found for given task.

Parameters:

task

Returns:

property is_multimode: bool
property max_number_of_mode: int
class discrete_optimization.generic_tasks_tools.multimode.SinglemodeSolution(problem: Problem)[source]

Bases: MultimodeSolution[Task]

get_mode(task: Task) → int[source]

Retrieve mode found for given task.

Parameters:

task

Returns:

problem: SinglemodeProblem[Task]

discrete_optimization.generic_tasks_tools.multimode_scheduling module

class discrete_optimization.generic_tasks_tools.multimode_scheduling.MultimodeSchedulingProblem[source]

Bases: SchedulingProblem[Task], MultimodeProblem[Task], Generic[Task]

Scheduling problem whose tasks durations depend only on mode.

This derives from problem with renewable resources, some of them are cumulative, some are not (e.g. unary resource if it is moreover an allocation problem). The task consumption of these cumulative resources is supposed to be determined entirely determined by the task mode.

abstractmethod get_task_mode_duration(task: Task, mode: int) → int[source]

Get task duration according to mode.

Parameters:

  • task

  • mode – not used for single-mode problems

Returns:

class discrete_optimization.generic_tasks_tools.multimode_scheduling.MultimodeSchedulingSolution(problem: Problem)[source]

Bases: SchedulingSolution[Task], MultimodeSolution[Task], Generic[Task]

Solution type associated to MultimodeSchedulingProblem.

check_duration_constraints() → bool[source]
check_task_duration_constraint(task: Task) → bool[source]
problem: MultimodeSchedulingProblem[Task]

discrete_optimization.generic_tasks_tools.precedence module

class discrete_optimization.generic_tasks_tools.precedence.PrecedenceProblem[source]

Bases: TasksProblem[HashableTask]

Problem with precedence constraints on tasks.

abstractmethod get_precedence_constraints() → dict[HashableTask, Iterable[HashableTask]][source]

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

get_precedence_graph() → Graph[source]

discrete_optimization.generic_tasks_tools.renewable_resource module

class discrete_optimization.generic_tasks_tools.renewable_resource.RenewableResourceProblem[source]

Bases: SchedulingProblem[Task], Generic[Task, Resource]

Base class for scheduling problems dealing with renewable resources whose availability depend on a calendar.

get_fake_tasks(resource: Resource) → list[tuple[int, int, int]][source]

Get fake tasks explaining the delta between resource current capacity and its max capacity

Parameters:

resource

Returns:

list of intervals of the form (start, end, value), for task starting at start and ending at end, using value resource.

abstractmethod get_resource_availabilities(resource: Resource) → 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_resource_calendar(resource: Resource, horizon: int | None = None) → list[int][source]

Compute resource calendar.

Parameters:

  • resource

  • horizon – max value of time considered. Default to self.get_makespan_upper_bound().

Returns:

list of resource value by time step from 0 to horizon-1.

get_resource_consolidated_availabilities(resource: Resource, horizon: int | None = None) → list[tuple[int, int, int]][source]

Get availabilities intervals for a given resource, consolidated as a partition of [0, horizon)

Default implementation use the get_calendar_resource_availabilities, by assuming the intervals are disjunct but potentially lacking 0-valued intervals.

Parameters:

  • resource

  • horizon – max value of time considered. Default to self.get_makespan_upper_bound().

Returns:

sorted 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_resource_max_capacity(resource: Resource) → int[source]

Get max capacity of the given resource

Default implementation take the max over its calendar and cache it.

Parameters:

resource

Returns:

abstract property renewable_resources_list: list[Resource]

Renewable resources used by the tasks.

Notes

  • renewable = the resource replenishes as soon as a task using it ends;

  • it can be a mix of unary resources (e.g. employees) and cumulative resources (e.g. tool types).

update_resource_availabilities() → None[source]

Method to call when the resource availabilities have changed.

Default implementation clears the cache on get_resource_max_capacity().

class discrete_optimization.generic_tasks_tools.renewable_resource.RenewableResourceSolution(problem: Problem)[source]

Bases: SchedulingSolution[Task], Generic[Task, Resource]

check_all_resource_capacity_constraints() → bool[source]

Check capacity constraint on all renewable resources.

check_resource_capacity_constraint(resource: Resource) → bool[source]

Check capacity constraint on given renewable resource.

check_resources_capacity_constraints(resources: Iterable[Resource]) → bool[source]

Check capacity constraint respected on given resources.

Do it simultaneously on all resources to optimize computation time.

abstractmethod get_resource_consumption(resource: Resource, task: Task) → int[source]

Get resource consumption by given task.

Parameters:

  • resource

  • task

Returns:

problem: RenewableResourceProblem[Task, Resource]
discrete_optimization.generic_tasks_tools.renewable_resource.consolidate_availability_intervals(intervals: list[tuple[int, int, int]], horizon: int)[source]

Ensure that the intervals are a partition of [0, horizon).

Parameters:

  • intervals – intervals (start, end, value). Supposed to be disjunct.

  • horizon – max value of time considered

Returns:

sorted list of intervals constituting a partition of [0, horizon)

discrete_optimization.generic_tasks_tools.renewable_resource.consolidate_calendar(calendar: list[int], horizon: int)[source]

Consoidate the calendar to correspond to the given horizon

If too long, retursn calendar[:horizon]. If too short, fill with 0’s.

Parameters:

  • calendar

  • horizon

Returns:

a calendar of size horizon

discrete_optimization.generic_tasks_tools.renewable_resource.convert_availability_intervals_to_calendar(intervals: list[tuple[int, int, int]], horizon: int) → list[int][source]

Convert availability intervals into a calendar.

Parameters:

  • intervals – availability intervals, assumed to be disjunct

  • horizon – maximum time step considered

Returns:

list of available resource values for each time step

discrete_optimization.generic_tasks_tools.renewable_resource.convert_calendar_to_availability_intervals(calendar: int | list[int], horizon: int) → list[tuple[int, int, int]][source]

Convert a calendar into availability intervals.

Parameters:

  • calendar – if integer means a constant value, else list of values for each time step. If len(calendar)<horizon, last values are assumed to be 0. If len(calendar)>horizon, it is truncated to calendar[:horizon]

  • horizon – maximum time step considered

Returns:

list of (start,end, value), a sorted partition of [0, horizon)

discrete_optimization.generic_tasks_tools.renewable_resource.merge_resources_availability_intervals(intervals_per_resource: list[list[tuple[int, int, int]]], horizon: int) → list[tuple[int, int, int]][source]

Merge several resources availability intervals, considering all resources as one meta-resource

Parameters:

  • intervals_per_resource – availability for each resource

  • horizon – maximum time step considered

Returns:

availability intervals for the meta-resource

discrete_optimization.generic_tasks_tools.renewable_resource.merge_resources_calendars(calendars: list[list[int]], horizon: int) → list[int][source]

Merge several resources calendars, considering all resources as one meta-resource

Parameters:

  • calendars – calendars for each resource to merge

  • horizon – maximum time step considered

Returns:

calendar for the meta-resource

discrete_optimization.generic_tasks_tools.scheduling module

class discrete_optimization.generic_tasks_tools.scheduling.SchedulingCpSolver(problem: Problem, params_objective_function: ParamsObjectiveFunction | None = None, **kwargs: Any)[source]

Bases: TasksCpSolver[Task]

Base class for cp solvers handling scheduling problems.

abstractmethod add_constraint_chaining_tasks(task1: Task, task2: Task) → list[Any][source]

Add constraint chaining task1 with task2

task2 start == task1 end

Parameters:

  • task1

  • task2

Returns:

resulting constraints

abstractmethod add_constraint_on_task(task: Task, start_or_end: StartOrEnd, sign: SignEnum, time: int) → list[Any][source]

Add constraint on given task start or end

task start or end must compare to time according to sign

Parameters:

  • task

  • start_or_end

  • sign

  • time

Returns:

resulting constraints

get_global_makespan_variable() → Any[source]

Construct and get the variable tracking the global makespan.

Default implementation uses get_subtasks_makespan_variable on last tasks. Beware: a further call to get_subtasks_makespan_variable with another subset of tasks can change the constraints on this variable and thus make it obsolete.

Returns:

objective variable to minimize

get_makespan_lower_bound() → int[source]

Get a lower bound on global makespan.

Can be overriden in solvers wanting to specify it in init_model() for instance.

get_makespan_upper_bound() → int[source]

Get a upper bound on global makespan.

abstractmethod get_subtasks_makespan_variable(subtasks: Iterable[Task]) → Any[source]

Construct and get the variable tracking the makespan on a subset of tasks.

Beware: a further call to get_subtasks_makespan_variable with another subset of tasks can change the constraints on this variable and thus make it obsolete.

Parameters:

subtasks

Returns:

objective variable to minimize

abstractmethod get_subtasks_sum_end_time_variable(subtasks: Iterable[Task]) → Any[source]

Construct and get the variable tracking the sum of end times on a subset of tasks.

Parameters:

subtasks

Returns:

objective variable to minimize

abstractmethod get_subtasks_sum_start_time_variable(subtasks: Iterable[Task]) → Any[source]

Construct and get the variable tracking the sum of start times on a subset of tasks.

Parameters:

subtasks

Returns:

objective variable to minimize

problem: SchedulingProblem[Task]
class discrete_optimization.generic_tasks_tools.scheduling.SchedulingProblem[source]

Bases: TasksProblem[Task]

Base class for scheduling problems.

A scheduling problems is about finding start and end times to tasks.

get_last_tasks() → list[Task][source]

Get a sublist of tasks that are candidate to be the last one scheduled.

Default to all tasks.

get_makespan_lower_bound() → int[source]

Get a lower bound on global makespan.

Default to 0. But can be overriden for problems with more information.

abstractmethod get_makespan_upper_bound() → int[source]

Get a upper bound on global makespan.

class discrete_optimization.generic_tasks_tools.scheduling.SchedulingSolution(problem: Problem)[source]

Bases: TasksSolution[Task]

Base class for solution to scheduling problems.

constraint_chaining_tasks_satisfied(task1: Task, task2: Task) → bool[source]
constraint_on_task_satisfied(task: Task, start_or_end: StartOrEnd, sign: SignEnum, time: int) → bool[source]
get_duration(task: Task) → int[source]
abstractmethod get_end_time(task: Task) → int[source]
get_max_end_time() → int[source]
get_running_tasks(time: int) → list[Task][source]

Extract tasks running at given time.

abstractmethod get_start_time(task: Task) → int[source]
problem: SchedulingProblem[Task]

Module contents