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

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 use problem.is_compatible_task_unary_resource().

But you can override it if you want to have more constraints in the solver than in the problem.

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.

compatible_unary_resources(task: Task) → set[UnaryResource][source]

get_index_from_unary_resource(unary_resource: UnaryResource) → int[source]

get_same_unary_allocation() → list[set[Task]][source]: a element (t1, t2, t3..) of this means corresponds to task for which we want the same resource to be allocated. WARNING: To be overriden if meaningful in the problem :return:

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_allocation_consistency() → bool[source]

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_allocated() → int[source]: Compute number of tasks with at least a resource allocated.

compute_nb_unary_resource_usages(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]

get_task_allocation(task: Task) → set[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:

is_present(task: Task) → bool[source]

Tell whether the task is present in the solution.

For allocation problem, default to “at least one unary resource has been allocated to the task”. To be overriden in child classes for problem having tasks present without allocation.

problem: AllocationProblem[Task, UnaryResource]

class discrete_optimization.generic_tasks_tools.allocation.WithoutAllocationProblem[source]

Bases: AllocationProblem[Task, None], Generic[Task]

Mixin to simplify deriving from GenericSchedulingProblem when no allocation is needed.

property unary_resources_list: list[None]

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.WithoutAllocationSolution(problem: Problem)[source]

Bases: AllocationSolution[Task, None], Generic[Task]

Mixin to simplify deriving from GenericSchedulingSolution when no allocation is needed.

check_allocation_consistency() → bool[source]

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:

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.alternative_subproblems module

class discrete_optimization.generic_tasks_tools.alternative_subproblems.AlternativeSchedulingProblem[source]

Bases: MultimodeSchedulingProblem[Task], Generic[Task]

get_alternative_scheduling_subproblem() → list[AlternativeSchedulingSubProblem][source]

class discrete_optimization.generic_tasks_tools.alternative_subproblems.AlternativeSchedulingSolution(problem: Problem)[source]

Bases: MultimodeSchedulingSolution[Task]

check_alternative_scheduling_subproblem() → None[source]

problem: AlternativeSchedulingProblem[Task]

class discrete_optimization.generic_tasks_tools.alternative_subproblems.AlternativeSchedulingSubProblem(source_task: Hashable, sink_task: Hashable, graph: networkx.classes.digraph.DiGraph | None = None, is_graph_successors: bool = True, list_paths: list[list[Hashable]] | None = None, is_path_successors: bool = True, nb_path_to_do: int = 1)[source]

Bases: object

graph: DiGraph | None = None

is_graph_successors: bool = True

is_path_successors: bool = True

list_paths: list[list[Hashable]] | None = None

nb_path_to_do: int = 1

sink_task: Hashable

source_task: Hashable

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.

get_index_from_task(task: Task) → int[source]

get_task_from_index(i: int) → Task[source]

has_optional_tasks() → bool[source]

is_optional(task: Task) → 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 optional_tasks_list: list[Task]

abstract property tasks_list: list[Task]: List of all tasks to schedule or allocate to.

update_tasks_list() → None[source]: To be call when tasks_list is updated to reset the cache.

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

Bases: Solution, Generic[Task]

Base class for scheduling/allocation solutions.

check_present_tasks() → bool[source]

get_present_tasks()[source]

is_present(task: Task) → 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: TasksProblem[Task]

discrete_optimization.generic_tasks_tools.calendar_resource module

class discrete_optimization.generic_tasks_tools.calendar_resource.CalendarResourceProblem[source]

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

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

abstract property calendar_resources_list: list[Resource]

Renewable resources with an availability calendar 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).
calendar can be constant

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:

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.calendar_resource.CalendarResourceSolution(problem: Problem)[source]

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

check_all_calendar_resource_capacity_constraints() → bool[source]: Check capacity constraint on all calendar resources.

check_calendar_resource_capacity_constraint(resource: Resource) → bool[source]: Check capacity constraint on given resource.

check_calendar_resource_capacity_constraints(resources: Iterable[Resource]) → bool[source]

Check capacity constraint respected on given resources.

Do it simultaneously on all resources to optimize computation time.

compute_aggregated_calendar_resources_levels(weights: dict[Resource, int] | None = None) → int[source]

Compute aggregated level (i.e. min capacity needed) of each calendar resource.

Parameters:: weights – optional weights to apply to each resource in the sum. Default to 1.

compute_calendar_resources_levels() → dict[Resource, int][source]: Compute the level (i.e. min capacity needed) for each calendar resource.

compute_nb_calendar_resources_used(weights: dict[Resource, int] | None = None) → int[source]

Compute number of calendar resources used by at least one task.

Parameters:: weights – optional weights to apply to each resource in the sum. Default to 1.

Returns:

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

Get resource consumption by given task.

Parameters:

resource
task

Returns:

problem: CalendarResourceProblem[Task, Resource]

class discrete_optimization.generic_tasks_tools.calendar_resource.WithoutCalendarResourceProblem[source]

Bases: CalendarResourceProblem[Task, None], Generic[Task]

property calendar_resources_list: list[Resource]

Renewable resources with an availability calendar 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).
calendar can be constant

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)

class discrete_optimization.generic_tasks_tools.calendar_resource.WithoutCalendarResourceSolution(problem: Problem)[source]

Bases: CalendarResourceSolution[Task, None], Generic[Task]

check_all_calendar_resource_capacity_constraints() → bool[source]: Check capacity constraint on all calendar resources.

check_calendar_resource_capacity_constraint(resource: Resource) → bool[source]: Check capacity constraint on given resource.

check_calendar_resource_capacity_constraints(resources: Iterable[Resource]) → bool[source]

Check capacity constraint respected on given resources.

Do it simultaneously on all resources to optimize computation time.

compute_aggregated_calendar_resources_levels(weights: dict[Resource, int] | None = None)[source]

Compute aggregated level (i.e. min capacity needed) of each calendar resource.

Parameters:: weights – optional weights to apply to each resource in the sum. Default to 1.

compute_calendar_resources_levels() → dict[Resource, int][source]: Compute the level (i.e. min capacity needed) for each calendar resource.

compute_nb_calendar_resources_used(weights: dict[Resource, int] | None = None) → int[source]

Compute number of calendar resources used by at least one task.

Parameters:: weights – optional weights to apply to each resource in the sum. Default to 1.

Returns:

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

Get resource consumption by given task.

Parameters:

resource
task

Returns:

discrete_optimization.generic_tasks_tools.calendar_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.calendar_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.calendar_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.calendar_resource.convert_calendar_to_availability_intervals(calendar: int | list[int] | NDArray[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.calendar_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.calendar_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.cumulative_resource module

class discrete_optimization.generic_tasks_tools.cumulative_resource.CumulativeResourceProblem[source]

Bases: CalendarResourceProblem[Task, CumulativeResource | OtherCalendarResource], MultimodeSchedulingProblem[Task], Generic[Task, CumulativeResource, OtherCalendarResource]

Scheduling problem with cumulative resources consumed by tasks.

Supports two consumption modes:

Standard: Task consumption is fixed by task mode. Example: Task A in mode 1 always consumes 5 units.
Resource-dependent: Task consumption depends on other tasks’ modes. Modeled via a consumption mapping. If the task/mode dont depend on any other task, returns empty condition with the static resource need.

Resource-Dependent Example:

Task A’s electricity varies based on whether task B’s heater is on:

>>> mapping = {
...     frozenset([("task_B", 0)]): 10,  # B heater off: A uses 10 kW
...     frozenset([("task_B", 1)]): 8,   # B heater on: A uses 8 kW
... }

Implementation pattern:

>>> class MyProblem(CumulativeResourceProblem):
...     def is_cumulative_resource_task_mode_consumption_dependent(
...         self, resource, task, mode
...     ):
...         return resource in self.mode_details[task][mode]
...
...     def get_cumulative_resource_consumption_mapping(
...         self, resource, task, mode
...     ):
...         if self.is_cumulative_resource_task_mode_consumption_dependent(
...             resource, task, mode
...         ):
...             return self.mode_details[task][mode][resource]
...         return None
...
...

See src/discrete_optimization/rcpsp_resource_dependent/problem.py for a complete implementation.

abstract property cumulative_resources_list: list[CumulativeResource]

abstractmethod get_cumulative_resource_consumption(resource: CumulativeResource, task: Task, 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_cumulative_resource_consumption_mapping(resource: CumulativeResource, task: Task, mode: int) → dict[frozenset[tuple[Task, int]], int] | None[source]

Get resource consumption mapping for resource-dependent tasks.

Parameters:

resource – The cumulative resource
task – The task
mode – The task mode

Returns:

Mapping from task/mode configurations to consumption values. Keys are frozensets of (task, mode) tuples. Returns None if task has standard (non-dependent) consumption.

Example

>>> mapping = problem.get_cumulative_resource_consumption_mapping(
...     "electricity", task_A, 0
... )
{frozenset([("task_B", 0)]): 100, frozenset([("task_B", 1)]): 80}

get_possible_cumulative_resource_consumption(resource: CumulativeResource, task: Task, mode: int) → set[int][source]

get_possible_cumulative_resource_consumption_all_modes(resource: CumulativeResource, task: Task) → set[int][source]

has_any_cumulative_consumption_dependent()[source]

is_cumulative_resource(resource: CumulativeResource | OtherCalendarResource) → bool[source]

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

Parameters:: resource

Returns:

is_cumulative_resource_task_consumption_dependent(resource: CumulativeResource, task: Task)[source]

is_cumulative_resource_task_mode_consumption_dependent(resource: CumulativeResource, task: Task, mode: int) → bool[source]

is_task_cumulative_consumption_dependent(task: Task)[source]

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

Bases: CalendarResourceSolution[Task, CumulativeResource | OtherCalendarResource], MultimodeSchedulingSolution[Task], Generic[Task, CumulativeResource, OtherCalendarResource]

Solution type associated to CumulativeResourceProblem.

get_calendar_resource_consumption(resource: CumulativeResource | OtherCalendarResource, task: Task) → int[source]

Get resource consumption by given task.

Dispatches to either: - get_cumulative_resource_consumption() for standard (fixed) consumption - get_calendar_resource_consumption_from_mapping() for resource-dependent consumption

Parameters:

resource – The calendar resource (must be cumulative)
task – The task

Returns:

Resource consumption amount

Raises:

NotImplementedError – If resource is not cumulative

get_calendar_resource_consumption_from_mapping(resource: CumulativeResource, task: Task) → int[source]

Retrieve resource consumption from mapping based on current solution’s mode assignments.

Parameters:

resource – The cumulative resource
task – The task

Returns:

Resource consumption value, or None if no matching configuration found.

Example

Given mapping {frozenset([(task_B, 1)]): 8, frozenset([(task_B, 0)]): 10} and solution with task_B in mode 1, returns 8.

problem: CumulativeResourceProblem[Task, CumulativeResource, OtherCalendarResource]

class discrete_optimization.generic_tasks_tools.cumulative_resource.WithoutCumulativeResourceProblem[source]

Bases: CumulativeResourceProblem[Task, None, OtherCalendarResource], Generic[Task, OtherCalendarResource]

Mixin for problem without cumulative resources.

To be used has an additional mixin with generic GenericSchedulingProblem.

property cumulative_resources_list: list[CumulativeResource]

get_cumulative_resource_consumption(resource: CumulativeResource, task: Task, 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.

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

Bases: CumulativeResourceSolution[Task, None, OtherCalendarResource], Generic[Task, OtherCalendarResource]

Mixin for solution without cumulative resources.

To be used has an additional mixin with generic GenericSchedulingSolution.

discrete_optimization.generic_tasks_tools.entities module

Entity abstraction for scheduling constraints.

This module provides a unified abstraction for scheduling entities that can participate in various types of constraints (precedence, time lags, resource blocking, etc.).

An entity represents any schedulable object with start and end times: - Individual tasks (TaskEntity) - Groups of tasks (GroupEntity) - Tasks in specific modes (TaskModeEntity) - Hierarchical compositions of entities (CompositeEntity)

The abstraction is recursive: CompositeEntity can contain any other entities (including other CompositeEntity instances), enabling arbitrary hierarchical structures for constraint modeling (e.g., projects → phases → tasks).

Entities are immutable (frozen dataclasses) so they can be used as dict keys.

class discrete_optimization.generic_tasks_tools.entities.CompositeEntity(entities: frozenset[SchedulingEntity], composite_id: Hashable | None = None)[source]

Bases: SchedulingEntity[Task]

Entity representing a hierarchical composition of other entities.

This enables recursive entity structures for hierarchical constraint modeling: - Groups of groups (e.g., projects containing sub-projects) - Mixed collections of tasks, groups, and mode-specific entities - Arbitrarily nested entity hierarchies (e.g., departments → teams → tasks)

The composite’s start time is the minimum start of its active children. The composite’s end time is the maximum end of its active children.

Examples

# Hierarchy: Project → Phases → Tasks phase1 = CompositeEntity(

entities=frozenset({TaskEntity(t1), TaskEntity(t2)}), composite_id=”phase1”

) phase2 = CompositeEntity(

entities=frozenset({TaskEntity(t3), TaskEntity(t4)}), composite_id=”phase2”

) project = CompositeEntity(

entities=frozenset({phase1, phase2}), composite_id=”project_alpha”

)

# Mixed: combining different entity types mixed = CompositeEntity(

entities=frozenset({
TaskEntity(t1), GroupEntity(tasks=frozenset({t2, t3})), TaskModeEntity(task=t4, mode=2)

})

)

entities

Set of child entities (must be non-empty)

Type:: frozenset[discrete_optimization.generic_tasks_tools.entities.SchedulingEntity]

composite_id

Optional identifier for display/debugging

Type:: collections.abc.Hashable | None

Raises:: ValueError – If entities set is empty, or if all children are inactive when querying start/end times

composite_id: Hashable | None = None

entities: frozenset[SchedulingEntity]

property entity_id: Hashable

Unique identifier for this entity.

Used for hashing, equality, and display.

Returns:: Hashable identifier (str, int, tuple, etc.)

get_end_time(solution: SchedulingSolution) → int | AbsentValue.ABSENT[source]

Get the end time of this entity in the given solution.

For tasks: the task’s end time For groups: maximum end time of tasks in group For conditional entities: end time if active, else raises error

Parameters:: solution – The scheduling solution to query
Returns:: End time (integer)
Raises:: ValueError – If entity is not active/present in solution

get_start_time(solution: SchedulingSolution) → int | AbsentValue.ABSENT[source]

Get the start time of this entity in the given solution.

For tasks: the task’s start time For groups: minimum start time of tasks in group For conditional entities: start time if active, else raises error

Parameters:: solution – The scheduling solution to query
Returns:: Start time (integer)
Raises:: ValueError – If entity is not active/present in solution (e.g., TaskModeEntity when task is not in the specified mode)

get_tasks() → frozenset[Task][source]

Get all tasks that compose this entity.

For single tasks: {task} For groups: all tasks in the group For conditional entities: {task}

Returns:: Frozen set of tasks

is_active(solution: SchedulingSolution) → bool[source]

Check if this entity is active/present in the solution.

For tasks: always True (task is always scheduled) For groups: True if any task in group is scheduled For conditional entities: True if the condition is satisfied

(e.g., TaskModeEntity is active only if task is in specified mode)

Parameters:: solution – The scheduling solution to query
Returns:: True if entity is active, False otherwise

class discrete_optimization.generic_tasks_tools.entities.GroupEntity(tasks: frozenset[Task], group_id: Hashable | None = None)[source]

Bases: SchedulingEntity[Task]

Entity representing a group/batch of tasks.

The group’s start time is the minimum start of its tasks. The group’s end time is the maximum end of its tasks.

This is useful for: - Representing projects with multiple tasks - Modeling batches that must stay together - Defining spans that consume resources

tasks

Set of tasks in the group (must be non-empty)

Type:: frozenset[discrete_optimization.generic_tasks_tools.base.Task]

group_id

Optional identifier for the group (for display/debugging)

Type:: collections.abc.Hashable | None

Examples

>>> entity = GroupEntity(
...     tasks=frozenset({"prep", "main", "cleanup"}),
...     group_id="maintenance_job_1"
... )
>>> entity.get_start_time(solution)  # min(start of prep, main, cleanup)
>>> entity.get_end_time(solution)    # max(end of prep, main, cleanup)
>>> entity.get_tasks()
frozenset({'prep', 'main', 'cleanup'})

property entity_id: Hashable

Unique identifier for this entity.

Used for hashing, equality, and display.

Returns:: Hashable identifier (str, int, tuple, etc.)

get_end_time(solution: SchedulingSolution) → int | AbsentValue[source]

Get the end time of this entity in the given solution.

For tasks: the task’s end time For groups: maximum end time of tasks in group For conditional entities: end time if active, else raises error

Parameters:: solution – The scheduling solution to query
Returns:: End time (integer)
Raises:: ValueError – If entity is not active/present in solution

get_start_time(solution: SchedulingSolution) → int | AbsentValue[source]

Get the start time of this entity in the given solution.

For tasks: the task’s start time For groups: minimum start time of tasks in group For conditional entities: start time if active, else raises error

Parameters:: solution – The scheduling solution to query
Returns:: Start time (integer)
Raises:: ValueError – If entity is not active/present in solution (e.g., TaskModeEntity when task is not in the specified mode)

get_tasks() → frozenset[Task][source]

Get all tasks that compose this entity.

For single tasks: {task} For groups: all tasks in the group For conditional entities: {task}

Returns:: Frozen set of tasks

group_id: Hashable | None = None

is_active(solution: SchedulingSolution) → bool[source]

Check if this entity is active/present in the solution.

For tasks: always True (task is always scheduled) For groups: True if any task in group is scheduled For conditional entities: True if the condition is satisfied

(e.g., TaskModeEntity is active only if task is in specified mode)

Parameters:: solution – The scheduling solution to query
Returns:: True if entity is active, False otherwise

tasks: frozenset[Task]

class discrete_optimization.generic_tasks_tools.entities.SchedulingEntity[source]

Bases: Generic[Task]

Abstract representation of a scheduling entity that has start and end times.

An entity represents any schedulable object that can participate in constraints: - Individual tasks (TaskEntity) - Groups of tasks (GroupEntity) - Tasks in specific execution modes (TaskModeEntity) - Hierarchical compositions of entities (CompositeEntity) - Other aggregations (resources, shifts, projects, etc.)

Entities are immutable (frozen dataclass) so they can be used as dict keys.

The entity abstraction is recursive: CompositeEntity can contain other entities, enabling hierarchical constraint modeling (e.g., projects → phases → tasks).

The entity abstraction allows expressing complex constraints naturally: - “Group of tasks must finish before another task starts” (precedence) - “Resource blocked from end of group to start of task” (resource blocking) - “If task is in mode 2, then block resource X” (conditional blocking) - “All phases of a project must respect a resource limit” (hierarchical constraints)

Example

>>> from discrete_optimization.generic_tasks_tools.entities import TaskEntity, GroupEntity
>>> # Individual task
>>> task_ent = TaskEntity("paint")
>>> # Group of tasks
>>> group_ent = GroupEntity(frozenset({"prep", "paint", "dry"}), "painting_job")
>>> # Query in solution
>>> start = task_ent.get_start_time(solution)
>>> group_start = group_ent.get_start_time(solution)  # min of task starts

abstract property entity_id: Hashable

Unique identifier for this entity.

Used for hashing, equality, and display.

Returns:: Hashable identifier (str, int, tuple, etc.)

abstractmethod get_end_time(solution: SchedulingSolution) → int | AbsentValue[source]

Get the end time of this entity in the given solution.

For tasks: the task’s end time For groups: maximum end time of tasks in group For conditional entities: end time if active, else raises error

Parameters:: solution – The scheduling solution to query
Returns:: End time (integer)
Raises:: ValueError – If entity is not active/present in solution

abstractmethod get_start_time(solution: SchedulingSolution) → int | AbsentValue[source]

Get the start time of this entity in the given solution.

For tasks: the task’s start time For groups: minimum start time of tasks in group For conditional entities: start time if active, else raises error

Parameters:: solution – The scheduling solution to query
Returns:: Start time (integer)
Raises:: ValueError – If entity is not active/present in solution (e.g., TaskModeEntity when task is not in the specified mode)

abstractmethod get_tasks() → frozenset[Task][source]

Get all tasks that compose this entity.

For single tasks: {task} For groups: all tasks in the group For conditional entities: {task}

Returns:: Frozen set of tasks

abstractmethod is_active(solution: SchedulingSolution) → bool[source]

Check if this entity is active/present in the solution.

For tasks: always True (task is always scheduled) For groups: True if any task in group is scheduled For conditional entities: True if the condition is satisfied

(e.g., TaskModeEntity is active only if task is in specified mode)

Parameters:: solution – The scheduling solution to query
Returns:: True if entity is active, False otherwise

class discrete_optimization.generic_tasks_tools.entities.TaskEntity(task: Task)[source]

Bases: SchedulingEntity[Task]

Entity representing a single task.

This is the most common entity type, wrapping a Task reference.

task

The task this entity represents

Type:: discrete_optimization.generic_tasks_tools.base.Task

Examples

>>> entity = TaskEntity(task="assembly")
>>> entity.get_start_time(solution)
>>> entity.get_end_time(solution)
>>> entity.get_tasks()
frozenset({'assembly'})
>>> entity.is_active(solution)

property entity_id: Hashable

Unique identifier for this entity.

Used for hashing, equality, and display.

Returns:: Hashable identifier (str, int, tuple, etc.)

get_end_time(solution: SchedulingSolution) → int | AbsentValue[source]

Get the end time of this entity in the given solution.

For tasks: the task’s end time For groups: maximum end time of tasks in group For conditional entities: end time if active, else raises error

Parameters:: solution – The scheduling solution to query
Returns:: End time (integer)
Raises:: ValueError – If entity is not active/present in solution

get_start_time(solution: SchedulingSolution) → int | AbsentValue[source]

Get the start time of this entity in the given solution.

For tasks: the task’s start time For groups: minimum start time of tasks in group For conditional entities: start time if active, else raises error

Parameters:: solution – The scheduling solution to query
Returns:: Start time (integer)
Raises:: ValueError – If entity is not active/present in solution (e.g., TaskModeEntity when task is not in the specified mode)

get_tasks() → frozenset[Task][source]

Get all tasks that compose this entity.

For single tasks: {task} For groups: all tasks in the group For conditional entities: {task}

Returns:: Frozen set of tasks

is_active(solution: SchedulingSolution) → bool[source]

Check if this entity is active/present in the solution.

For tasks: always True (task is always scheduled) For groups: True if any task in group is scheduled For conditional entities: True if the condition is satisfied

(e.g., TaskModeEntity is active only if task is in specified mode)

Parameters:: solution – The scheduling solution to query
Returns:: True if entity is active, False otherwise

task: Task

class discrete_optimization.generic_tasks_tools.entities.TaskModeEntity(task: Task, mode: int)[source]

Bases: SchedulingEntity[Task]

Entity representing a task in a specific execution mode.

This entity is only “active” if the task is executed in the specified mode. Useful for mode-dependent constraints: - “If task A is in mode 1, then block resource X” - “Task B can only start after task A in mode 2 completes”

task

The task

Type:: discrete_optimization.generic_tasks_tools.base.Task

mode

The specific mode (integer)

Type:: int

Examples

>>> entity = TaskModeEntity(task="painting", mode=2)
>>> entity.is_active(solution)  # True only if painting is in mode 2
False  # (if painting is in mode 1)
>>> # If active:
>>> entity.get_start_time(solution)
10
>>> entity.get_tasks()
frozenset({'painting'})

Raises:: ValueError – When calling get_start_time() or get_end_time() on an inactive entity

property entity_id: Hashable

Unique identifier for this entity.

Used for hashing, equality, and display.

Returns:: Hashable identifier (str, int, tuple, etc.)

get_end_time(solution: SchedulingSolution) → int | AbsentValue.ABSENT[source]

Get the end time of this entity in the given solution.

For tasks: the task’s end time For groups: maximum end time of tasks in group For conditional entities: end time if active, else raises error

Parameters:: solution – The scheduling solution to query
Returns:: End time (integer)
Raises:: ValueError – If entity is not active/present in solution

get_start_time(solution: SchedulingSolution) → int | AbsentValue.ABSENT[source]

Get the start time of this entity in the given solution.

For tasks: the task’s start time For groups: minimum start time of tasks in group For conditional entities: start time if active, else raises error

Parameters:: solution – The scheduling solution to query
Returns:: Start time (integer)
Raises:: ValueError – If entity is not active/present in solution (e.g., TaskModeEntity when task is not in the specified mode)

get_tasks() → frozenset[Task][source]

Get all tasks that compose this entity.

For single tasks: {task} For groups: all tasks in the group For conditional entities: {task}

Returns:: Frozen set of tasks

is_active(solution: SchedulingSolution) → bool[source]: Check if the task is executed in the specified mode.

mode: int

task: Task

discrete_optimization.generic_tasks_tools.enums module

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

Bases: Enum

ABSENT = 'absent'

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

Bases: Enum

MAX = 'max'

MIN = 'min'

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

Bases: Enum

END = 'end'

START = 'start'

discrete_optimization.generic_tasks_tools.generic_scheduling module

class discrete_optimization.generic_tasks_tools.generic_scheduling.GenericSchedulingProblem[source]

Bases: ResourceBlockingProblem[Task, Skill | NonSkillCumulativeResource, UnaryResource], SkillProblem[Task, UnaryResource, Skill, NonSkillCumulativeResource, UnaryResource], NonRenewableResourceProblem[Task, NonRenewableResource], PrecedenceSchedulingProblem[Task], TimelagProblem[Task], TimewindowProblem[Task], NoOverlapProblem[Task], AlternativeSchedulingProblem[Task], Generic[Task, UnaryResource, Skill, NonSkillCumulativeResource, NonRenewableResource]

Scheduling problem with all optional features

This class derives from other mixins to provide utilities that require that mix: - scheduling: tasks need to be scheduled - calendar: the renewable resources have their own calendar that will be used for constraining allocations and schedule - multimode: the tasks have several mode on which the duration depends - cumulative: the tasks consume cumulative resources according to the chosen mode - allocation: the tasks can have unary resources allocated to them - skill: some cumulative resource are skills that are brought to tasks by allocated unary resources - non-renewable: the tasks consume non-renewable resources according to the chosen mode - precedence: precedence constraints between tasks - resource blocking: resources blocked during non-execution periods (gaps, spans) - cost: the choice of a mode or of an allocation has a given cost

Even though this class is generic but encompasses also more specific cases: - singlemode: actually only one mode per task - no skills: if skills_list is empty - no allocation: unary_resources is empty - no cumulative ressources: if resources_list list only unary resources - no calendar: resource capacity can be given as a constant on [0, horizon) - no non-renewable ressources: if non_renewable_resources_list empty - no precedence constraints: precedence constraints empty - no resource blocking: no blocking constraints defined - no cost: cost = 0

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

This generic class is to be used to construct generic automatic solvers (e.g. ).

property calendar_resources_list: list[Skill | NonSkillCumulativeResource | UnaryResource]

Renewable resources with an availability calendar 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).
calendar can be constant

check_calendar_resources_list() → None[source]

Check calendar resources list.

Raises:: AssertionError – if duplicates appear in the list

Returns:

compute_tighter_task_bounds(use_cpm: bool = False, horizon: int | None = None) → dict[Task, tuple[int, int, int, int]][source]

Compute tighter task bounds from problem time windows and min-max tak durations.

Parameters:

use_cpm – whether to use CPM propagating bounds through precedence graph
horizon – new horizon to take into account when computing tighter bounds, default to problem horizon.

Returns:

(start_lower_bound, end_lower_bound, start_upper_bound, end_upper_bound)}

Return type:

{task

evaluate(variable: GenericSchedulingSolution) → 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_consolidated_precedence_constraints() → dict[Task, set[Task]][source]

Consolidate precedence constraints defined by problem.

It takes into account time lags constraints. - end to start min constraint with non-negative offsets => precedence constraint - start synchronization => corresponding tasks should appear together in successors - end synchornization => corresponding tasks should share their successors

get_consolidated_time_lags(task1_start_or_end: StartOrEnd, task2_start_or_end: StartOrEnd, min_or_max: MinOrMax)[source]

Get consolidated time lags.

Same normalization as in TimelagProblem parent class. Also taking into account precedence constraints to enrich end to start min time lags.

Parameters:

task1_start_or_end
task2_start_or_end
min_or_max

Returns:

abstractmethod get_list_objective_computer() → list[ObjectiveComputer][source]

get_makespan_lower_bound() → int[source]

Get a lower bound on global makespan.

Computed tighter lower bounds on last tasks can be used to get a better makespan lower bound.

get_makespan_tighter_lower_bound(use_cpm: bool = False, horizon: int | None = None) → int[source]

Get a tighter lower bound on global makespan.

Parameters:

use_cpm – whether to use CPM bound propagation through precedence graph to improve tightness
horizon – new horizon to take into account when computing tighter bounds, default to problem horizon. NB: The choice of horizon should not affect the result for the lower bound, but it could avoid CPM complete recomputation thanks to caching if it was already launched with same horizon.

get_makespan_tighter_upper_bound() → int[source]

Compute a tighter upper bound on makespan.

The original makespan upper bound is used when computing tighter bounds for tasks starts and ends, via self.compute_tighter_task_bounds() or self.get_task_start_or_end_tighter_upper_bound(). From that tighter bounds, we can derive a new makespan upper bound.

get_objective_computer(objective: Objective) → list[ObjectiveComputer] | None[source]

get_objective_register() → ObjectiveRegister[source]

Returns the objective definition.

Returns (ObjectiveRegister): object defining the objective criteria.

get_task_start_or_end_tighter_lower_bound(task: Task, start_or_end: StartOrEnd, use_cpm: bool = False, horizon: int | None = None) → int[source]

Get a tighter lower bound on task start or end using possible durations.

Parameters:

use_cpm – whether to use CPM propagating bounds through precedence graph
horizon – new horizon to take into account when computing tighter bounds, default to problem horizon.

get_task_start_or_end_tighter_upper_bound(task: Task, start_or_end: StartOrEnd, use_cpm: bool = False, horizon: int | None = None) → int[source]

Get a tighter upper bound on task start or end using possible durations.

Parameters:

use_cpm – whether to use CPM propagating bounds through precedence graph
horizon – new horizon to take into account when computing tighter bounds, default to problem horizon.

is_unary_resource(resource: Skill | NonSkillCumulativeResource | UnaryResource) → bool[source]: Check if given resource is a unary resource.

satisfy(variable: GenericSchedulingSolution) → 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_partial(variable: GenericSchedulingSolution, duration: bool = True, calendar: bool = True, non_renewable_capacity: bool = True, precedence: bool = True, skill: bool = True, allocation: bool = True, time_lags: bool = True, time_windows: bool = True, no_overlap: bool = True, forbidden_intervals: bool = True, resource_blocking: bool = True, mode_constraints: bool = True, optional_tasks: bool = True, alternative_scheduling: bool = True) → bool[source]

Partial checks on solution.

One can switch off some checks by setting the corresponding parameter to False.

Parameters:

variable
duration
calendar
non_renewable_capacity
precedence
skill
allocation
time_lags
time_windows
no_overlap
forbidden_intervals
resource_blocking
mode_constraints
optional_tasks
alternative_scheduling

Returns:

update_precedence_constraints() → None[source]

Method to call when precedence constraints have been updated.

Clear cache from consolidated precedence constraints and time lags.

Returns:

update_resource_availabilities() → None[source]

Method to call when the resource availabilities have changed.

Default implementation clears the cache on get_resource_max_capacity().

update_task_bounds() → None[source]

Method to be called when problem time windows are updated.

It clears necessary cache on computed tighter bounds.

update_time_lags() → None[source]

Method to call when time lags have been updated.

Clear cache from consolidated precedence constraints and time lags.

Returns:

class discrete_optimization.generic_tasks_tools.generic_scheduling.GenericSchedulingSolution(problem: Problem)[source]

Bases: ResourceBlockingSolution[Task, Skill | NonSkillCumulativeResource, UnaryResource], SkillSolution[Task, UnaryResource, Skill, NonSkillCumulativeResource, UnaryResource], NonRenewableResourceSolution[Task, NonRenewableResource], PrecedenceSchedulingSolution[Task], TimelagSolution[Task], TimewindowSolution[Task], NoOverlapSolution[Task], AlternativeSchedulingSolution[Task], Generic[Task, UnaryResource, Skill, NonSkillCumulativeResource, NonRenewableResource]

Solution type associated to GenericSchedulingProblem.

compute_workload_dispersion() → int[source]

get_calendar_resource_consumption(resource: Skill | NonSkillCumulativeResource | UnaryResource, task: Task) → int[source]

is_present(task: Task) → 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: GenericSchedulingProblem[Task, UnaryResource, Skill, NonSkillCumulativeResource, NonRenewableResource]

discrete_optimization.generic_tasks_tools.generic_scheduling_impl module

class discrete_optimization.generic_tasks_tools.generic_scheduling_impl.GenericSchedulingImplProblem(horizon: int, durations_per_mode: dict[~collections.abc.Hashable, dict[int, int]], resource_consumptions: dict[~collections.abc.Hashable, dict[int, dict[~collections.abc.Hashable, int]]] = <factory>, resource_consumptions_dependent: dict[~collections.abc.Hashable, dict[int, dict[~collections.abc.Hashable, dict[frozenset[tuple[~collections.abc.Hashable, int]], int]]]] = <factory>, successors: dict[~collections.abc.Hashable, set[~collections.abc.Hashable]] = <factory>, unary_resources: set[~collections.abc.Hashable] = <factory>, unary_resources_skills: dict[~collections.abc.Hashable, dict[~collections.abc.Hashable, int]] = <factory>, unary_resources_availabilities: dict[~collections.abc.Hashable, list[tuple[int, int]]] = <factory>, unary_resources_task_compatibility: dict[~collections.abc.Hashable, set[~collections.abc.Hashable]] = <factory>, skills: set[~collections.abc.Hashable] = <factory>, non_skill_cumulative_resources: dict[~collections.abc.Hashable, int | list[tuple[int, int, int]]] = <factory>, non_renewable_resources: dict[~collections.abc.Hashable, int] = <factory>, time_windows: dict[~collections.abc.Hashable, tuple[int | None, int | None, int | None, int | None]] = <factory>, start_to_start_min_time_lags: list[tuple[~collections.abc.Hashable, ~collections.abc.Hashable, int]] = <factory>, start_to_end_min_time_lags: list[tuple[~collections.abc.Hashable, ~collections.abc.Hashable, int]] = <factory>, end_to_start_min_time_lags: list[tuple[~collections.abc.Hashable, ~collections.abc.Hashable, int]] = <factory>, end_to_end_min_time_lags: list[tuple[~collections.abc.Hashable, ~collections.abc.Hashable, int]] = <factory>, no_overlap_sets: set[frozenset[~collections.abc.Hashable]] = <factory>, forbidden_intervals: dict[~collections.abc.Hashable, list[tuple[int, int]]] = <factory>, flexible_gap_blocking_constraints: list[tuple[~discrete_optimization.generic_tasks_tools.entities.SchedulingEntity, ~discrete_optimization.generic_tasks_tools.enums.StartOrEnd, ~discrete_optimization.generic_tasks_tools.entities.SchedulingEntity, ~discrete_optimization.generic_tasks_tools.enums.StartOrEnd, dict[~collections.abc.Hashable, int], ~discrete_optimization.generic_tasks_tools.resource_blocking.BlockingConstraintMetadata]] = <factory>, span_blocking_constraints: list[tuple[~discrete_optimization.generic_tasks_tools.entities.SchedulingEntity, dict[~collections.abc.Hashable, int], ~discrete_optimization.generic_tasks_tools.resource_blocking.BlockingConstraintMetadata]] = <factory>, mode_constraints: list[tuple[~discrete_optimization.generic_tasks_tools.multimode.ModeConstraintType, list[tuple[~collections.abc.Hashable, int]]]] = <factory>, same_unary_allocation: list[set[~collections.abc.Hashable]] = <factory>, objective: dataclasses.InitVar[discrete_optimization.generic_tasks_tools.generic_scheduling_utils.Objective | collections.abc.Iterable[tuple[discrete_optimization.generic_tasks_tools.generic_scheduling_utils.Objective, int]]] = Objective.MAKESPAN, custom_evaluate_fn: ~typing.Callable[[~discrete_optimization.generic_tasks_tools.generic_scheduling_impl.GenericSchedulingImplSolution], int] | None = None, objective_resource_weights: dict[~collections.abc.Hashable, int] | None = None, mode_costs: dict[~collections.abc.Hashable, dict[int, int]] = <factory>, unary_resource_costs: dict[~collections.abc.Hashable, dict[int, dict[~collections.abc.Hashable, int]]] = <factory>, compute_time_penalty: bool = True, optional_tasks: set[~collections.abc.Hashable] = <factory>, list_objective_computer: list[~discrete_optimization.generic_tasks_tools.objectives.objective_computer.ObjectiveComputer] = None, alternative_scheduling_subproblems: list[~discrete_optimization.generic_tasks_tools.alternative_subproblems.AlternativeSchedulingSubProblem] = <factory>)[source]

Bases: GenericSchedulingProblem[Hashable, Hashable, Hashable, Hashable, Hashable]

Generic implementation of a scheduling problem.

It implements the abstract class GenericSchedulingProblem.

horizon

max allowed time to finish the tasks

Type:: int

durations_per_mode

task -> mode -> duration. Tasks durations, mode by mode. This is used to know all available tasks, all available modes for a given task, and corresponding durations.

Type:: dict[collections.abc.Hashable, dict[int, int]]

resource_consumptions

task -> mode -> resource -> conso. Cumulative or non-renewable resource consumption, task by task, mode by mode. The resource can be a skill. Missing key => conso = 0

Type:: dict[collections.abc.Hashable, dict[int, dict[collections.abc.Hashable, int]]]

successors

maps a task to its successors in the precedence graph. Each successor task must start after the given task ends. Default to no precedence constraints. Note that a consolidated version of it will be constructed using the time lags constraints.

Type:: dict[collections.abc.Hashable, set[collections.abc.Hashable]]

unary_resources

available unary resources. Default to none.

Type:: set[collections.abc.Hashable]

unary_resources_skills

skill values of each unary resource. Missing key => skill value = 0

Type:: dict[collections.abc.Hashable, dict[collections.abc.Hashable, int]]

unary_resources_availabilities

availability of unary resources on the form of list of intervals (start, end) Missing key => always available.

Type:: dict[collections.abc.Hashable, list[tuple[int, int]]]

unary_resources_task_compatibility

maps a task to its compatible unary resources. Missing key => all unary resources allowed.

Type:: dict[collections.abc.Hashable, set[collections.abc.Hashable]]

skills

available skills

Type:: set[collections.abc.Hashable]

non_skill_cumulative_resources

cumulative resources (excluding skills) availabilities. Format: either int => always available at the given max capacity, or list of intervals + capacity (start, end, value)

Type:: dict[collections.abc.Hashable, int | list[tuple[int, int, int]]]

non_renewable_resources

non-renewable resources max capacities

Type:: dict[collections.abc.Hashable, int]

time_windows

maps task to start_lb, end_lb, start_ub, end_ub s.t. start_lb <= start(task) <= start_ub and end_lb <= end(task) <= end_ub missing or none value means 0 (lb) or self.horizon (ub)

Type:: dict[collections.abc.Hashable, tuple[int | None, int | None, int | None, int | None]]

start_to_start_min_time_lags

min time lags constraints between task starts. task1, task2, offset meaning start(task1) + offset <= start(task2) Note that using negative offset can model start-to-start max time lags.

Type:: list[tuple[collections.abc.Hashable, collections.abc.Hashable, int]]

start_to_end_min_time_lags

min time lags constraints first task start and second task end. task1, task2, offset meaning start(task1) + offset <= end(task2) Note that using negative offset can model end-to-start max time lags.

Type:: list[tuple[collections.abc.Hashable, collections.abc.Hashable, int]]

end_to_start_min_time_lags

min time lags constraints between first task end and second task start. task1, task2, offset meaning end(task1) + offset <= start(task2) Note that using negative offset can model start-to-end max time lags.

Type:: list[tuple[collections.abc.Hashable, collections.abc.Hashable, int]]

end_to_end_min_time_lags

min time lags constraints between task ends. task1, task2, offset meaning end(task1) + offset <= end(task2) Note that using negative offset can model end-to-end max time lags.

Type:: list[tuple[collections.abc.Hashable, collections.abc.Hashable, int]]

no_overlap_sets

a set of (set of tasks that should not overlap together)

Type:: set[frozenset[collections.abc.Hashable]]

forbidden_intervals

maps task to forbidden intervals that cannot overlap with it. Missing key => no forbidden intervals.

Type:: dict[collections.abc.Hashable, list[tuple[int, int]]]

flexible_gap_blocking_constraints

list of blocking constraints between entity points. Each constraint is (entity1, point1, entity2, point2, resources, metadata). Default to no blocking constraints.

Type:: list[tuple[discrete_optimization.generic_tasks_tools.entities.SchedulingEntity, discrete_optimization.generic_tasks_tools.enums.StartOrEnd, discrete_optimization.generic_tasks_tools.entities.SchedulingEntity, discrete_optimization.generic_tasks_tools.enums.StartOrEnd, dict[collections.abc.Hashable, int], discrete_optimization.generic_tasks_tools.resource_blocking.BlockingConstraintMetadata]]

span_blocking_constraints

list of span blocking constraints. Each constraint is (tasks, resources, metadata) where tasks is a frozenset. Default to no blocking constraints.

Type:: list[tuple[discrete_optimization.generic_tasks_tools.entities.SchedulingEntity, dict[collections.abc.Hashable, int], discrete_optimization.generic_tasks_tools.resource_blocking.BlockingConstraintMetadata]]

objective

objective for the problem. Default to minimization of makespan. Either an iterable of (objective, weight) so that the problem should maximize the aggregated objective resulting from weighted sum of objectives, or a single objective in which case we use the corresponding default weight from discrete_optimization.generic_tasks_tools.generic_scheduling_utils.OBJECTIVE_DEFAULT_WEIGHTS and maximize it. For instance the default weight for makespan is -1 so that it will actually minimize the makespan.

Type:: dataclasses.InitVar[discrete_optimization.generic_tasks_tools.generic_scheduling_utils.Objective | collections.abc.Iterable[tuple[discrete_optimization.generic_tasks_tools.generic_scheduling_utils.Objective, int]]]

custom_evaluate_fn

function used to evaluate the “custom” objective (to be maximized).

Type:: Callable[[discrete_optimization.generic_tasks_tools.generic_scheduling_impl.GenericSchedulingImplSolution], int] | None

objective_resource_weights

Weights to be used by the objective when summing used resources (Objective.NB_RESOURCES_USED) or resources levels (Objective.RESOURCES_LEVELS). Default to 1 for resources not mentioned.

Type:: dict[collections.abc.Hashable, int] | None

mode_costs

cost of choosing each mode. Missing key => cost = 0.

Type:: dict[collections.abc.Hashable, dict[int, int]]

unary_resource_costs

cost of allocating each unary resource. Missing key => cost = 0.

Type:: dict[collections.abc.Hashable, dict[int, dict[collections.abc.Hashable, int]]]

compute_time_penalty

whether to include time penalties in evaluation

Type:: bool

alternative_scheduling_subproblems: list[AlternativeSchedulingSubProblem]

check_resources_lists() → None[source]: Check duplicates in resources.

compute_time_penalty: bool = True

create_subproblem_from_partial_solution(partial_solution: RawSolution[Hashable, Hashable, Hashable]) → GenericSchedulingImplProblem[source]

Create a subproblem according to a partial solution.

Tasks already scheduled are removed from subproblem.
Add time windows constraints to model timelags/precedence constraints with removed tasks.
Update resource calendars with scheduled allocations
Update non-renewable resources max capacities
Transform no_overlap constraints into forbidden intervals constraints

custom_evaluate_fn: Callable[[GenericSchedulingImplSolution], int] | None = None

durations_per_mode: dict[Hashable, dict[int, int]]

end_to_end_min_time_lags: list[tuple[Hashable, Hashable, int]]

end_to_start_min_time_lags: list[tuple[Hashable, Hashable, int]]

evaluate(variable: GenericSchedulingSolution) → 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.

flexible_gap_blocking_constraints: list[tuple[SchedulingEntity, StartOrEnd, SchedulingEntity, StartOrEnd, dict[Hashable, int], BlockingConstraintMetadata]]

forbidden_intervals: dict[Hashable, list[tuple[int, int]]]

get_alternative_scheduling_subproblem() → list[AlternativeSchedulingSubProblem][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_cumulative_resource_consumption(resource: Hashable, 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_cumulative_resource_consumption_mapping(resource: Hashable, task: Hashable, mode: int) → dict[frozenset[tuple[Hashable, int]], int][source]

Get resource consumption mapping for resource-dependent tasks.

Parameters:

resource – The cumulative resource
task – The task
mode – The task mode

Returns:

Mapping from task/mode configurations to consumption values. Keys are frozensets of (task, mode) tuples. Returns None if task has standard (non-dependent) consumption.

Example

>>> mapping = problem.get_cumulative_resource_consumption_mapping(
...     "electricity", task_A, 0
... )
{frozenset([("task_B", 0)]): 100, frozenset([("task_B", 1)]): 80}

get_dummy_solution() → Solution[source]

Create a trivial solution for the problem.

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

get_end_to_end_min_time_lags() → list[tuple[Hashable, Hashable, int]][source]

Get min time lags between task ends.

Default to no min time lags. Should be overriden in child class for problems with min time lags.

Returns:: list of task1, task2, offset meaning end(task1) + offset <= end(task2)

get_end_to_start_min_time_lags() → list[tuple[Hashable, Hashable, int]][source]

Get min time lags between first task end and second task start.

Default to no min time lags. Should be overriden in child class for problems with min time lags.

Returns:: list of task1, task2, offset meaning end(task1) + offset <= start(task2)

get_flexible_gap_blocking_constraints() → list[tuple[SchedulingEntity, StartOrEnd, SchedulingEntity, StartOrEnd, dict[Hashable, int], BlockingConstraintMetadata]][source]: Return flexible gap blocking constraints.

get_forbidden_intervals(task: Hashable) → list[tuple[int, int]][source]

Get fixed intervals that should not overlap with given task.

Default to empty list. To be overridden in child classes.

Parameters:: task
Returns:: List of intervals (start, end) with start <= end (will not be checked)

get_list_objective_computer() → list[ObjectiveComputer][source]

get_makespan_upper_bound() → int[source]: Get an upper bound on global makespan.

get_mode_constraints() → list[tuple[ModeConstraintType, list[tuple[Hashable, int]]]][source]

An element of the list is a tuple of (ModeConstraintType, list of (task,mode)) that implies the other choice of mode. For example (SORTED_IMPLICATION, [(T1, 1), (T2, 2), (T3, 1)]) means : if T1 is in mode 1, T2 is in mode 2, T3 is in mode 1.. This can be useful to model mode choice that has an influence on the future mode choice. For example, in an assembly line if we choose a given station path for a product, it should stay on it ! if mode_constraint_type == ModeConstraintType.SORTED_IMPLICATION:

then the constraint is not active only when T1 is in mode 1, it should be true if any of the task,mode is active. So if mode(T2)==2 then the other mode are also forced!

Returns:

get_no_overlap() → set[frozenset[Hashable]][source]

An object in this returned set is a (frozen) set of task, where no task should overlap with another one in this set.

Default to no such sets.

get_non_renewable_resource_capacity(resource: Hashable) → int[source]

Get resource max capacity

Parameters:: resource

Returns:

get_non_renewable_resource_consumption(resource: Hashable, task: Hashable, mode: int) → int[source]

Get resource consumption of the task in the given mode

Parameters:

resource – non-renewable resource
task
mode – not used for single mode problems

Returns:.

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

get_non_renewable_resource_consumption_mapping(resource: Hashable, task: Hashable, mode: int) → dict[frozenset[tuple[Hashable, int]], int][source]

get_precedence_constraints() → dict[Hashable, Iterable[Hashable]][source]: Map each task to the tasks that need to be performed after it.

get_resource_availabilities(resource: Hashable) → 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]: a element (t1, t2, t3..) of this means corresponds to task for which we want the same resource to be allocated. WARNING: To be overriden if meaningful in the problem :return:

get_solution_type() → type[Solution][source]

Returns the class implementation of a Solution.

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

get_span_blocking_constraints() → list[tuple[SchedulingEntity, dict[Hashable, int], BlockingConstraintMetadata]][source]: Return span blocking constraints.

get_start_to_end_min_time_lags() → list[tuple[Hashable, Hashable, int]][source]

Get min time lags between first task start and second task end.

Default to no min time lags. Should be overriden in child class for problems with min time lags.

Returns:: list of task1, task2, offset meaning start(task1) + offset <= end(task2)

get_start_to_start_min_time_lags() → list[tuple[Hashable, Hashable, int]][source]

Get min time lags between tasks starts.

Default to no min time lags. Should be overriden in child class for problems with min time lags.

Returns:: list of task1, task2, offset meaning start(task1) + offset <= start(task2)

get_task_mode_duration(task: Hashable, mode: int) → int[source]

Get task duration according to mode.

Parameters:

task
mode – not used for single-mode problems

Returns:

get_task_modes(task: Hashable) → set[int][source]

Retrieve mode found for given task.

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_unary_resource_skill_value(unary_resource: Hashable, skill: Hashable) → int[source]: Skill value of given resource for given skill.

horizon: int

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_cumulative_resource_task_mode_consumption_dependent(resource: Hashable, task: Hashable, mode: int) → bool[source]

is_non_renewable_resource_task_mode_consumption_dependent(resource: Hashable, task: Hashable, mode: int) → bool[source]

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.

list_objective_computer: list[ObjectiveComputer] = None

mode_constraints: list[tuple[ModeConstraintType, list[tuple[Hashable, int]]]]

mode_costs: dict[Hashable, dict[int, int]]

no_overlap_sets: set[frozenset[Hashable]]

non_renewable_resources: dict[Hashable, int]

property non_renewable_resources_list: list[Hashable]: Non-renewable resources used by the tasks.

non_skill_cumulative_resources: dict[Hashable, int | list[tuple[int, int, int]]]

property non_skill_cumulative_resources_list: list[Hashable]: List of cumulative resources that are not skills.

objective: dataclasses.InitVar[discrete_optimization.generic_tasks_tools.generic_scheduling_utils.Objective | collections.abc.Iterable[tuple[discrete_optimization.generic_tasks_tools.generic_scheduling_utils.Objective, int]]] = 'makespan'

objective_resource_weights: dict[Hashable, int] | None = None

optional_tasks: set[Hashable]

resource_consumptions: dict[Hashable, dict[int, dict[Hashable, int]]]

resource_consumptions_dependent: dict[Hashable, dict[int, dict[Hashable, dict[frozenset[tuple[Hashable, int]], int]]]]

same_unary_allocation: list[set[Hashable]]

set_fixed_attributes(attribute_name: str, solution: Solution) → 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:

skills: set[Hashable]

property skills_list: list[Hashable]: List of skills needed by tasks and brought by unary resources.

span_blocking_constraints: list[tuple[SchedulingEntity, dict[Hashable, int], BlockingConstraintMetadata]]

start_to_end_min_time_lags: list[tuple[Hashable, Hashable, int]]

start_to_start_min_time_lags: list[tuple[Hashable, Hashable, int]]

successors: dict[Hashable, set[Hashable]]

property tasks_list: list[Hashable]: List of all tasks to schedule or allocate to.

time_windows: dict[Hashable, tuple[int | None, int | None, int | None, int | None]]

unary_resource_costs: dict[Hashable, dict[int, dict[Hashable, int]]]

unary_resources: set[Hashable]

unary_resources_availabilities: dict[Hashable, list[tuple[int, int]]]

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.

unary_resources_skills: dict[Hashable, dict[Hashable, int]]

unary_resources_task_compatibility: dict[Hashable, set[Hashable]]

update_problem()[source]: Method to call when some attributes of the problem are modified.

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.generic_scheduling_impl.GenericSchedulingImplSolution(problem: GenericSchedulingImplProblem, raw_sol: RawSolution[Hashable, Hashable, Hashable])[source]

Bases: GenericSchedulingSolution[Hashable, Hashable, Hashable, Hashable, Hashable]

Generic implementation of a solution to a scheduling problem.

It implements the abstract class GenericSchedulingSolution.

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

Retrieve mode found for given task.

Hypothesis:: The returned mode 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:

get_task_allocation(task: Hashable) → set[Hashable][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:

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.

is_skill_used(task: Hashable, unary_resource: Hashable, skill: Hashable) → bool[source]

Tell whether the given skill from given unary_resource is used in given task.

If True, self.is_allocated(task, unary_resource) must also be True. If the skill is not needed by the task or not in unary_resource skills, should return False.

lazy_copy() → Solution[source]

This function should return a new object but possibly with mutable attributes from the original objects.

A typical use of lazy copy is in evolutionary algorithms or genetic algorithm where the use of local move don’t need to do a possibly costly deepcopy.

Returns (Solution): copy (possibly shallow) of the Solution

problem: GenericSchedulingImplProblem

discrete_optimization.generic_tasks_tools.generic_scheduling_utils module

discrete_optimization.generic_tasks_tools.generic_scheduling_utils.OBJECTIVE_DEFAULT_WEIGHTS: dict[Objective, int] = {Objective.CUSTOM: 1, Objective.MAKESPAN: 1, Objective.NB_TASKS_ALLOCATED: -1, Objective.NB_UNARY_RESOURCES_USED: 1}: Default weight applied to a given objective so that it will be maximized.

class discrete_optimization.generic_tasks_tools.generic_scheduling_utils.Objective(*values)[source]

Bases: Enum

Objective for a generic scheduling problem.

ALLOCATION_CHANGES = 'allocation_changes'

ALLOCATION_COST = 'allocation_cost'

CALENDAR_RESOURCES_LEVELS = 'calendar_resources_levels': Weighted sum of resources levels (i.e. needed capacities), to minimize.

CUMUL_COST = 'cumulative_cost'

CUSTOM = 'custom_objective'

EARLINESS_TARDINESS = 'earliness_tardiness'

MAKESPAN = 'makespan': Global makespan of the schedule, to minimize.

MODE_COST = 'mode_cost'

NB_TASKS_ALLOCATED = 'nb_tasks_allocated': Number of tasks with at least one resource allocated, to maximize.

NB_TASKS_SCHEDULED = 'nb_tasks_scheduled': Number of tasks actually scheduled, to maximize (usually).

NB_UNARY_RESOURCES_USED = 'nb_unary_resources_used': Number of allocated unary resources, to minimize.

NON_RENEWABLE_RESOURCES_LEVELS = 'non_renewable_resources_levels': Weighted sum of non-renewable resources levels (i.e. needed capacities), to minimize.

SCHEDULE_CHANGES = 'scheduling_changes'

TIME_PENALTY = 'time_penalty'

discrete_optimization.generic_tasks_tools.generic_scheduling_utils.PENALTY_DEFAULT_WEIGHTS: dict[Penalty, int] = {Penalty.TIME: -100}: Default weight applied to a given penalty to be added to the objective so that it will be maximized.

class discrete_optimization.generic_tasks_tools.generic_scheduling_utils.Penalty(*values)[source]

Bases: Enum

Penalties for a generic scheduling problem.

TIME = 'time_penalty'

class discrete_optimization.generic_tasks_tools.generic_scheduling_utils.RawSolution(task_variables: dict[~discrete_optimization.generic_tasks_tools.base.Task, ~discrete_optimization.generic_tasks_tools.generic_scheduling_utils.TaskVariable[~discrete_optimization.generic_tasks_tools.allocation.UnaryResource, ~discrete_optimization.generic_tasks_tools.skill.Skill]], metadata: dict[str, ~typing.Any] = <factory>)[source]

Bases: Generic[Task, UnaryResource, Skill]

Raw format for a generic scheduling solution

Does not inherit from d-o Solution class.

You can do raw_sol_1 | raw_sol_2, it will return another raw solution merging both task_variables dictionaries, but dropping metadata.

.

metadata: dict[str, Any]

take_subset(tasks: Container[Task]) → RawSolution[Task, UnaryResource, Skill][source]

Take a subset of the solution by keeping only variables associated to given tasks

Parameters:: tasks – subset of tasks to keep
Returns:: The raw solution with variables for given tasks. Any metadata is dropped.

task_variables: dict[Task, TaskVariable[UnaryResource, Skill]]

class discrete_optimization.generic_tasks_tools.generic_scheduling_utils.TaskVariable(start: int | AbsentValue, end: int | AbsentValue, mode: int | AbsentValue, is_present: bool = True, allocated: dict[~discrete_optimization.generic_tasks_tools.allocation.UnaryResource, set[~discrete_optimization.generic_tasks_tools.skill.Skill]]=<factory>, info: dict[str, ~typing.Any]=<factory>)[source]

Bases: Generic[UnaryResource, Skill]

Task characteristics found in a generic scheduling solution.

allocated: dict[UnaryResource, set[Skill]]

end: int | AbsentValue

get_start_or_end(start_or_end: StartOrEnd) → int | AbsentValue[source]

info: dict[str, Any]

is_present: bool = True

mode: int | AbsentValue

start: int | AbsentValue

discrete_optimization.generic_tasks_tools.multimode module

class discrete_optimization.generic_tasks_tools.multimode.ModeConstraintType(*values)[source]

Bases: Enum

SORTED_IMPLICATION = 0

UNORDERED = 1

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.

get_mode_constraints() → list[tuple[ModeConstraintType, list[tuple[Task, int]]]][source]

An element of the list is a tuple of (ModeConstraintType, list of (task,mode)) that implies the other choice of mode. For example (SORTED_IMPLICATION, [(T1, 1), (T2, 2), (T3, 1)]) means : if T1 is in mode 1, T2 is in mode 2, T3 is in mode 1.. This can be useful to model mode choice that has an influence on the future mode choice. For example, in an assembly line if we choose a given station path for a product, it should stay on it ! if mode_constraint_type == ModeConstraintType.SORTED_IMPLICATION:

then the constraint is not active only when T1 is in mode 1, it should be true if any of the task,mode is active. So if mode(T2)==2 then the other mode are also forced!

Returns:

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.

check_mode_constraint() → bool[source]

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

Retrieve mode found for given task.

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

Parameters:: task

Returns:

has_a_mode(task: Task) → bool[source]

is_present(task: Task) → 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: 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.

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

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.

get_max_duration_of_tasks()[source]

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

class discrete_optimization.generic_tasks_tools.multimode_scheduling.SinglemodeSchedulingProblem[source]

Bases: SinglemodeProblem[Task], MultimodeSchedulingProblem[Task]

Single mode scheduling problems with fixed task durations.

Utility class simplifying MultimodeSchedulingProblem when single mode only.

abstractmethod get_task_duration(task: Task) → int[source]

Get task duration.

Parameters:: task

Returns:

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.SinglemodeSchedulingSolution(problem: Problem)[source]

Bases: SinglemodeSolution[Task], MultimodeSchedulingSolution[Task]

Solution for single mode scheduling problem with fixed task durations.

Utility class useful when needing to derive from GenericSchedulingSolution without multi mode to be able to use cpsat auto solver.

discrete_optimization.generic_tasks_tools.no_overlap module

class discrete_optimization.generic_tasks_tools.no_overlap.NoOverlapProblem[source]

Bases: SchedulingProblem[Task]

Problem with no overlap between tasks: For example for open-shop problem, where the order of each operations is arbitrary, but still no overlap.

get_forbidden_intervals(task: Task) → list[tuple[int, int]][source]

Get fixed intervals that should not overlap with given task.

Default to empty list. To be overridden in child classes.

Parameters:: task
Returns:: List of intervals (start, end) with start <= end (will not be checked)

get_no_overlap() → set[frozenset[Task]][source]

An object in this returned set is a (frozen) set of task, where no task should overlap with another one in this set.

Default to no such sets.

is_task_in_nooverlap_constraint(task: Task) → bool[source]

Tell if given task is part of a no overlapping constraint.

Etiher in a set returned by self.get_no_overlap() or with non-empty self.get_forbidden_intervals(task).

class discrete_optimization.generic_tasks_tools.no_overlap.NoOverlapSolution(problem: Problem)[source]

Bases: SchedulingSolution[Task]

Solution for problem with precedence constraints.

check_forbidden_intervals() → bool[source]

check_no_overlap() → bool[source]

problem: NoOverlapProblem[Task]

discrete_optimization.generic_tasks_tools.non_renewable_resource module

class discrete_optimization.generic_tasks_tools.non_renewable_resource.NonRenewableResourceProblem[source]

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

Base class for problems dealing with non-renewable resources consumed by tasks. Just like CumulativeResourceProblem, it supports two consumption modes:

Standard: Task consumption is fixed by task mode. Example: Task A in mode 1 always consumes 5 units.
Resource-dependent: Task consumption depends on other tasks’ modes. Modeled via a consumption mapping. If the task/mode dont depend on any other task, returns empty condition with the static resource need.

abstractmethod get_non_renewable_resource_capacity(resource: NonRenewableResource) → int[source]

Get resource max capacity

Parameters:: resource

Returns:

abstractmethod get_non_renewable_resource_consumption(resource: NonRenewableResource, task: Task, mode: int) → int[source]

Get resource consumption of the task in the given mode

Parameters:

resource – non-renewable resource
task
mode – not used for single mode problems

Returns:.

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

get_non_renewable_resource_consumption_mapping(resource: NonRenewableResource, task: Task, mode: int) → dict[frozenset[tuple[Task, int]], int][source]

get_possible_non_renewable_resource_consumption(resource: NonRenewableResource, task: Task, mode: int) → set[int][source]

get_possible_non_renewable_resource_consumption_all_modes(resource: NonRenewableResource, task: Task) → set[int][source]

has_any_non_renewable_consumption_dependent()[source]

is_non_renewable_resource_task_consumption_dependent(resource: NonRenewableResource, task: Task)[source]

is_non_renewable_resource_task_mode_consumption_dependent(resource: NonRenewableResource, task: Task, mode: int) → bool[source]

is_task_non_renewable_consumption_dependent(task: Task)[source]

abstract property non_renewable_resources_list: list[NonRenewableResource]: Non-renewable resources used by the tasks.

class discrete_optimization.generic_tasks_tools.non_renewable_resource.NonRenewableResourceSolution(problem: Problem)[source]

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

check_all_non_renewable_resource_capacity_constraints() → bool[source]: Check capacity constraint on all renewable resources.

check_non_renewable_resource_capacity_constraint(resource: NonRenewableResource) → bool[source]: Check capacity constraint on given renewable resource.

check_non_renewable_resource_capacity_constraints(resources: Iterable[NonRenewableResource])[source]

compute_aggregated_non_renewable_resources_consumptions(weights: dict[NonRenewableResource, int] | None = None)[source]

Compute aggregated consumption of each non-renewable resource by the solution.

Parameters:: weights – optional weights to apply to each resource in the sum. Default to 1.

compute_nb_non_renewable_resources_used(weights: dict[NonRenewableResource, int] | None = None) → int[source]

Compute number of non-renewable resources used by at least one task.

Parameters:: weights – optional weights to apply to each resource in the sum. Default to 1.

Returns:

compute_non_renewable_resources_consumptions() → dict[NonRenewableResource, int][source]: Compute total consumption of each non-renewable resource by the solution.

get_non_renewable_resource_consumption(resource: NonRenewableResource, task: Task) → int[source]

Get resource consumption by given task.

Parameters:

resource
task

Returns:

get_non_renewable_resource_consumption_from_mapping(resource: NonRenewableResource, task: Task) → int[source]

problem: NonRenewableResourceProblem[Task, NonRenewableResource]

class discrete_optimization.generic_tasks_tools.non_renewable_resource.WithoutNonRenewableResourceProblem[source]

Bases: NonRenewableResourceProblem[Task, None], Generic[Task]

Mixin for problem without non-renewable resources.

To be used has an additional mixin with generic GenericSchedulingProblem.

get_non_renewable_resource_capacity(resource: NonRenewableResource) → int[source]

Get resource max capacity

Parameters:: resource

Returns:

get_non_renewable_resource_consumption(resource: NonRenewableResource, task: Task, mode: int) → int[source]

Get resource consumption of the task in the given mode

Parameters:

resource – non-renewable resource
task
mode – not used for single mode problems

Returns:.

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

property non_renewable_resources_list: list[NonRenewableResource]: Non-renewable resources used by the tasks.

class discrete_optimization.generic_tasks_tools.non_renewable_resource.WithoutNonRenewableResourceSolution(problem: Problem)[source]

Bases: NonRenewableResourceSolution[Task, None], Generic[Task]

Mixin for solution without non-renewable resources.

To be used has an additional mixin with generic GenericSchedulingSolution.

check_all_non_renewable_resource_capacity_constraints() → bool[source]: Check capacity constraint on all renewable resources.

check_non_renewable_resource_capacity_constraint(resource: NonRenewableResource) → bool[source]: Check capacity constraint on given renewable resource.

check_non_renewable_resource_capacity_constraints(resources: Iterable[NonRenewableResource])[source]

compute_aggregated_non_renewable_resources_consumptions(weights: dict[NonRenewableResource, int] | None = None)[source]

Compute aggregated consumption of each non-renewable resource by the solution.

Parameters:: weights – optional weights to apply to each resource in the sum. Default to 1.

compute_nb_non_renewable_resources_used(weights: dict[NonRenewableResource, int] | None = None) → int[source]

Compute number of non-renewable resources used by at least one task.

Parameters:: weights – optional weights to apply to each resource in the sum. Default to 1.

Returns:

compute_non_renewable_resources_consumptions() → dict[NonRenewableResource, int][source]: Compute total consumption of each non-renewable resource by the solution.

discrete_optimization.generic_tasks_tools.plot_utils module

discrete_optimization.generic_tasks_tools.plot_utils.compute_nice_resource_consumption(scheduling_problem: GenericSchedulingProblem, scheduling_sol: GenericSchedulingSolution, list_resources: list[str] | None = None) → tuple[dict[int, NDArray[int64]], dict[int, NDArray[int64]]][source]

discrete_optimization.generic_tasks_tools.plot_utils.compute_resource_consumption(scheduling_problem: GenericSchedulingProblem, scheduling_sol: GenericSchedulingSolution, list_resources: list[str] | None = None, future_view: bool = True) → tuple[NDArray[int64], NDArray[int64]][source]

discrete_optimization.generic_tasks_tools.plot_utils.plot_ressource_view(scheduling_problem: GenericSchedulingProblem, scheduling_sol: GenericSchedulingSolution, list_resource: list[str] | None = None, title_figure: str = '', x_lim: list[int] | None = None, fig: Figure | None = None, ax: NDArray[object_] | None = None) → Figure[source]

discrete_optimization.generic_tasks_tools.plot_utils.plot_task_gantt(scheduling_problem: GenericSchedulingProblem, scheduling_sol: GenericSchedulingSolution, fig: Figure | None = None, ax: Axes | None = None, x_lim: list[int] | None = None, title: str | None = None) → Figure[source]

discrete_optimization.generic_tasks_tools.precedence module

class discrete_optimization.generic_tasks_tools.precedence.PrecedenceProblem[source]

Bases: TasksProblem[Task]

Problem with precedence constraints on tasks.

get_consolidated_precedence_constraints() → dict[Task, set[Task]][source]

Consolidate precedence constraints defined by problem.

In GenericSchedulingProblem it will also take into account time lags constraints. Default implementation is only taking get_precededence_constraints, removing the potential duplicates.

abstractmethod get_precedence_constraints() → dict[Task, Iterable[Task]][source]: Map each task to the tasks that need to be performed after it.

get_precedence_graph() → Graph[source]

update_precedence_constraints() → None[source]

Method to call when precedence constraints have been updated.

To be overriden in child classes.

Returns:

class discrete_optimization.generic_tasks_tools.precedence.PrecedenceSolution(problem: Problem)[source]

Bases: TasksSolution[Task]

Solution for problem with precedence constraints.

check_precedence_constraints() → bool[source]

Check that all precedence constraints are satisfied.

Returns:

abstractmethod check_tasks_order(task1, task2) → bool[source]

Check whether task1 is performed before task2.

Parameters:

task1
task2

Returns:

True if task1 is finished before task2 starts, False else.

problem: PrecedenceProblem[Task]

class discrete_optimization.generic_tasks_tools.precedence.WithoutPrecedenceProblem[source]

Bases: PrecedenceProblem[Task]

Utility mixin for problem w/o precedence constraints.

To be used has an additional mixin with generic GenericSchedulingProblem.

get_precedence_constraints() → dict[Task, Iterable[Task]][source]: Map each task to the tasks that need to be performed after it.

class discrete_optimization.generic_tasks_tools.precedence.WithoutPrecedenceSolution(problem: Problem)[source]

Bases: PrecedenceSolution[Task]

check_precedence_constraints() → bool[source]

Check that all precedence constraints are satisfied.

Returns:

discrete_optimization.generic_tasks_tools.precedence_scheduling module

class discrete_optimization.generic_tasks_tools.precedence_scheduling.PrecedenceSchedulingProblem[source]

Bases: PrecedenceProblem[Task], SchedulingProblem[Task]

Scheduling problem with precedence constraints on 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.

class discrete_optimization.generic_tasks_tools.precedence_scheduling.PrecedenceSchedulingSolution(problem: Problem)[source]

Bases: PrecedenceSolution[Task], SchedulingSolution[Task]

Solution for scheduling problem with precedence constraints.

Can implement check_tasks_order by using start and end times.

check_tasks_order(task1, task2) → bool[source]

Check whether task1 is performed before task2.

Parameters:

task1
task2

Returns:

True if task1 is finished before task2 starts, False else.

problem: PrecedenceSchedulingProblem[Task]

discrete_optimization.generic_tasks_tools.resource_blocking module

Resource blocking constraints for scheduling problems.

This module provides mixins for modeling resource blocking during non-execution periods: - Gap blocking: Resources blocked between two entities (e.g., changeover time) - Span blocking: Resources blocked for entire span of task group (e.g., project reservation)

Key features: - Flexible blocking points: START/END of entities (tasks, groups, conditional) - Calendar awareness: RESERVATION (spans unavailable periods) vs ACTIVE (must be available) - Overlap handling: Strategies to avoid double-counting when tasks overlap with blocking

class discrete_optimization.generic_tasks_tools.resource_blocking.BlockingConstraintMetadata(mode: BlockingMode = BlockingMode.RESERVATION, description: str = '')[source]

Bases: object

Metadata for a blocking constraint.

Resource blocking intervals are ALWAYS ADDITIVE with task consumption. The blocking demand is added on top of task consumption during the blocking period. Users should adjust their blocking demands accordingly: - If a task uses 2 units and blocking adds 1 unit, total consumption = 3 units - Ensure resource capacity can accommodate task + blocking consumption

mode

Calendar awareness mode controlling interaction with resource availability: - RESERVATION: Blocking can span unavailable periods (nights, weekends).

Resource is reserved even when “OFF”. Enforced without calendar constraints.

ACTIVE: Blocking only during available periods. Resource must be “ON”.
Enforced with calendar constraints.

Type:: discrete_optimization.generic_tasks_tools.resource_blocking.BlockingMode

description

Optional human-readable description of the constraint

Type:: str

description: str = ''

mode: BlockingMode = 'reservation'

class discrete_optimization.generic_tasks_tools.resource_blocking.BlockingMode(*values)[source]

Bases: Enum

Mode for resource blocking behavior.

RESERVATION: Resource slot is reserved but doesn’t need to be “ON” or available. Blocking can span periods when resource is unavailable (nights, weekends).

ACTIVE: Resource must be available/ON during blocking period. Blocking invalid if resource unavailable during any part of the period.

ACTIVE = 'active'

RESERVATION = 'reservation'

class discrete_optimization.generic_tasks_tools.resource_blocking.ResourceBlockingProblem[source]

Bases: CumulativeResourceProblem[Task, CumulativeResource, OtherCalendarResource], Generic[Task, CumulativeResource, OtherCalendarResource]

Mixin for problems with resource blocking constraints.

This mixin adds support for two types of blocking: 1. Flexible gap blocking: Block resources from one entity point to another 2. Span blocking: Block resources for entire span of task group

The problem should also inherit from CumulativeResourceProblem to provide resource definitions and capacity constraints.

get_flexible_gap_blocking_constraints() → list[tuple[SchedulingEntity, StartOrEnd, SchedulingEntity, StartOrEnd, dict[CumulativeResource, int], BlockingConstraintMetadata]][source]

Return flexible gap blocking constraints.

Each constraint blocks resources from one entity point to another entity point. Supports four patterns based on start/end combinations: - END → START: Classic gap/changeover (most common) - START → START: Preparation period - START → END: Full span coverage - END → END: Extended cleanup

Returns:

entity1: First entity (task, group, or conditional)
point1: START or END of first entity
entity2: Second entity
point2: START or END of second entity
resources: Dict mapping resources to consumption amounts
metadata: Blocking behavior configuration

Return type:

List of tuples (entity1, point1, entity2, point2, resources, metadata)

Default to no flexible gap blocking constraints.

get_span_blocking_constraints() → list[tuple[SchedulingEntity, dict[CumulativeResource, int], BlockingConstraintMetadata]][source]

Return span blocking constraints.

Each constraint blocks resources for the entire span of a task group: - From: minimum start time of any task in group - To: maximum end time of any task in group

Returns:

tasks: Frozen set of tasks defining the span
resources: Dict mapping resources to consumption amounts
metadata: Blocking behavior configuration

Return type:

List of tuples (tasks, resources, metadata)

Default to no span blocking constraints.

class discrete_optimization.generic_tasks_tools.resource_blocking.ResourceBlockingSolution(problem: Problem)[source]

Bases: CumulativeResourceSolution[Task, CumulativeResource, OtherCalendarResource], Generic[Task, CumulativeResource, OtherCalendarResource]

Mixin for solutions to problems with resource blocking constraints.

Provides methods to: - Compute resource consumption from blocking constraints - Check constraint satisfaction with calendar awareness - Handle overlap between blocking and task execution

Should be mixed with SchedulingSolution subclass. Inherits from MultimodeSolution to ensure get_mode() is always available.

check_blocking_constraints() → bool[source]

Check if all blocking constraints are satisfied.

Mirrors the two-constraint approach from CP-SAT solver:

Check 1 (RESERVATION constraint - no calendar):

Tasks + ALL blocking (RESERVATION + ACTIVE) <= base capacity
This allows RESERVATION blocking to span unavailable periods

Check 2 (ACTIVE constraint - with calendar):

Tasks + ACTIVE blocking <= calendar capacity at each time
ACTIVE mode: Validate blocking only occurs during available periods

Returns:: True if all constraints satisfied, False otherwise

compute_blocking_consumption(horizon: int, resource: CumulativeResource) → ndarray[source]

Compute resource consumption from all blocking constraints.

Blocking is always ADDITIVE: consumption from blocking is added to task consumption. This means total resource usage = task consumption + blocking consumption.

This method: 1. Computes blocking periods from flexible gap and span constraints 2. Adds blocking consumption for each period (ADDITIVE behavior) 3. Validates ACTIVE mode constraints against resource calendar

Parameters:

horizon – Time horizon for the schedule
resource – The resource to compute consumption for

Returns:

Array of length horizon with blocking consumption at each time point. Note: This is blocking consumption only. Task consumption is computed separately. Total consumption should be verified: task_consumption + blocking_consumption <= capacity

Raises:

ValueError – If ACTIVE mode blocking spans resource unavailable period

problem: ResourceBlockingProblem[Task, CumulativeResource, OtherCalendarResource]

satisfy() → bool[source]

Check if solution satisfies all constraints including blocking.

This extends the base satisfy() method to include blocking constraints.

Returns:: True if all constraints satisfied, False otherwise

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 an 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 | AbsentValue[source]

abstractmethod get_end_time(task: Task) → 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_max_end_time() → int[source]

Get makespan of the solution.

If no tasks are done, will return 0.

Returns:

get_running_tasks(time: int) → list[Task][source]

Extract tasks running at given time.

Optional tasks not “present” (according to is_present()) are removed from the list.

get_start_or_end_time(task: Task, start_or_end: StartOrEnd) → int | AbsentValue[source]

Get the start or end time for a given task.

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

abstractmethod get_start_time(task: Task) → 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_present(task: Task) → 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.

is_scheduled(task: Task) → bool[source]: Tell whether the task been scheduled.

problem: SchedulingProblem[Task]

discrete_optimization.generic_tasks_tools.skill module

Module containing mixins for skills.

A skill is a cumulative resource which is attached to a unary resource.

class discrete_optimization.generic_tasks_tools.skill.SkillProblem[source]

Bases: CumulativeResourceProblem[Task, Skill | NonSkillCumulativeResource, OtherCalendarResource], AllocationProblem[Task, UnaryResource], Generic[Task, UnaryResource, Skill, NonSkillCumulativeResource, OtherCalendarResource]

compute_skill_availabilities(skill: Skill) → list[tuple[int, int, int]][source]: Deduce skill availabilities from unary_resource availabilities and skill values.

property cumulative_resources_list: list[Skill | NonSkillCumulativeResource]

get_skills_of_task(task: Task) → set[Skill][source]

get_skills_of_unary_resource(unary_resource: UnaryResource) → set[Skill][source]

abstractmethod get_unary_resource_skill_value(unary_resource: UnaryResource, skill: Skill) → int[source]: Skill value of given resource for given skill.

get_unary_resource_with_skill(skill: Skill) → set[UnaryResource][source]

abstract property non_skill_cumulative_resources_list: list[Skill]: List of cumulative resources that are not skills.

only_one_skill_per_task: bool = False: Only one skill from each unary resource allocated to a given task can be used for the task.

abstract property skills_list: list[Skill]: List of skills needed by tasks and brought by unary resources.

update_skills()[source]

class discrete_optimization.generic_tasks_tools.skill.SkillSolution(problem: Problem)[source]

Bases: CumulativeResourceSolution[Task, Skill | NonSkillCumulativeResource, OtherCalendarResource], AllocationSolution[Task, UnaryResource], Generic[Task, UnaryResource, Skill, NonSkillCumulativeResource, OtherCalendarResource]

check_only_one_skill_per_task_and_unary_resource()[source]

check_skill_constraint(task: Task, skill: Skill, exact: bool = False, slack: int = 0) → bool[source]

check_skill_constraints(exact: bool = False, slack: int = 0) → bool[source]

check_skill_usage_and_allocation_consistency() → bool[source]

get_skill_value_on_task(task: Task, skill: Skill) → int[source]

abstractmethod is_skill_used(task: Task, unary_resource: UnaryResource, skill: Skill) → bool[source]

Tell whether the given skill from given unary_resource is used in given task.

If True, self.is_allocated(task, unary_resource) must also be True. If the skill is not needed by the task or not in unary_resource skills, should return False.

problem: SkillProblem[Task, UnaryResource, Skill, NonSkillCumulativeResource, OtherCalendarResource]

class discrete_optimization.generic_tasks_tools.skill.WithoutSkillProblem[source]

Bases: SkillProblem[Task, UnaryResource, None, NonSkillCumulativeResource, OtherCalendarResource], Generic[Task, UnaryResource, NonSkillCumulativeResource, OtherCalendarResource]

get_unary_resource_skill_value(unary_resource: UnaryResource, skill: Skill) → int[source]: Skill value of given resource for given skill.

property skills_list: list[Skill]: List of skills needed by tasks and brought by unary resources.

class discrete_optimization.generic_tasks_tools.skill.WithoutSkillSolution(problem: Problem)[source]

Bases: SkillSolution[Task, UnaryResource, None, NonSkillCumulativeResource, OtherCalendarResource], Generic[Task, UnaryResource, NonSkillCumulativeResource, OtherCalendarResource]

check_skill_constraint(task: Task, skill: None, exact: bool = False, slack: int = 0) → bool[source]

check_skill_constraints(exact: bool = False, slack: int = 0) → bool[source]

check_skill_usage_and_allocation_consistency() → bool[source]

is_skill_used(task: Task, unary_resource: UnaryResource, skill: Skill) → bool[source]

Tell whether the given skill from given unary_resource is used in given task.

If True, self.is_allocated(task, unary_resource) must also be True. If the skill is not needed by the task or not in unary_resource skills, should return False.

discrete_optimization.generic_tasks_tools.solvers_map module

discrete_optimization.generic_tasks_tools.solvers_map.get_solver_default_arguments(method: type[SolverDO]) → dict[str, Any][source]

discrete_optimization.generic_tasks_tools.solvers_map.look_for_solver(domain: Problem) → list[type[SolverDO]][source]

discrete_optimization.generic_tasks_tools.solvers_map.look_for_solver_class(class_domain: type[Problem]) → list[type[SolverDO]][source]

discrete_optimization.generic_tasks_tools.solvers_map.return_solver(method: type[SolverDO], problem: Problem, **kwargs: Any) → SolverDO[source]

discrete_optimization.generic_tasks_tools.solvers_map.solve(method: type[SolverDO], problem: Problem, **kwargs: Any) → ResultStorage[source]

discrete_optimization.generic_tasks_tools.solvers_map.solve_return_solver(method: type[SolverDO], problem: Problem, **kwargs: Any) → tuple[ResultStorage, SolverDO][source]

discrete_optimization.generic_tasks_tools.timelag module

class discrete_optimization.generic_tasks_tools.timelag.TimelagProblem[source]

Bases: SchedulingProblem[Task], Generic[Task]

Class for problem having time lags between tasks.

get_consolidated_time_lags(task1_start_or_end: StartOrEnd, task2_start_or_end: StartOrEnd, min_or_max: MinOrMax) → list[tuple[Task, Task, int]][source]

Get consolidated time lags.

The goal is to normalize the time lags list to avoid having duplicates or constraints implied by others. Note that it encompasses - same tuples in one of the lists - (t1, t2, offset) in end_to_start_min_time_lags and (t2, t1, -offset) in start_to_end_max_time_lags - several offsets for the same tasks (t1, t2)

The choices made here to normalize are: 1. keep only positive offsets in max time lags (the others are converted to min time lags) 2. keep only non-negative offsets in min time lags (the others are converted to max time lags) 3. take max(offsets) in min time lags when several offsets are available for the same tasks (t1, t2) 4. take min(offsets) in max time lags when several offsets are available for the same tasks (t1, t2) 5. drop (t1, t2, offset) in max time lag (with offset>0) if (t2, t1, offset’) with offset’>=0 is already in corresponding min time lag

as the latter implies trivially the other

Note that points 1, 4, and 5 amounts to taking min(offsets) on all tuples in max time lags + converted ones from min time lags (t2,t1 -offset), whichever the sign, and drop it if this min(offsets) is non-negative (whhich corresponds to the existence of (t2, t1, offset’) with offset’>=0 in min time lags.

Moreover this method makes the mapping between all methods according to start/end/min/max.

Parameters:

task1_start_or_end
task2_start_or_end
min_or_max

Returns:

get_end_to_end_max_time_lags() → list[tuple[Task, Task, int]][source]

Get max time lags between task ends.

Default to no max time lags. Should be overriden in child class for problems with max time lags.

Returns:: list of task1, task2, offset meaning end(task1) + offset >= end(task2)

get_end_to_end_min_time_lags() → list[tuple[Task, Task, int]][source]

Get min time lags between task ends.

Default to no min time lags. Should be overriden in child class for problems with min time lags.

Returns:: list of task1, task2, offset meaning end(task1) + offset <= end(task2)

get_end_to_start_max_time_lags() → list[tuple[Task, Task, int]][source]

Get max time lags between first task end and second task start.

Default to no max time lags. Should be overriden in child class for problems with max time lags.

Returns:: list of task1, task2, offset meaning end(task1) + offset >= start(task2)

get_end_to_start_min_time_lags() → list[tuple[Task, Task, int]][source]

Get min time lags between first task end and second task start.

Default to no min time lags. Should be overriden in child class for problems with min time lags.

Returns:: list of task1, task2, offset meaning end(task1) + offset <= start(task2)

get_original_time_lags(task1_start_or_end: StartOrEnd, task2_start_or_end: StartOrEnd, min_or_max: MinOrMax) → list[tuple[Task, Task, int]][source]

Maps strt/end/min/max choices to corresponding list of timelags

Parameters:

task1_start_or_end
task2_start_or_end
min_or_max

Returns:

get_start_to_end_max_time_lags() → list[tuple[Task, Task, int]][source]

Get max time lags between first task start and second task end.

Default to no max time lags. Should be overriden in child class for problems with max time lags.

Returns:: list of task1, task2, offset meaning start(task1) + offset >= end(task2)

get_start_to_end_min_time_lags() → list[tuple[Task, Task, int]][source]

Get min time lags between first task start and second task end.

Default to no min time lags. Should be overriden in child class for problems with min time lags.

Returns:: list of task1, task2, offset meaning start(task1) + offset <= end(task2)

get_start_to_start_max_time_lags() → list[tuple[Task, Task, int]][source]

Get max time lags between tasks starts.

Default to no max time lags. Should be overriden in child class for problems with max time lags.

Returns:: list of task1, task2, offset meaning start(task1) + offset >= start(task2)

get_start_to_start_min_time_lags() → list[tuple[Task, Task, int]][source]

Get min time lags between tasks starts.

Default to no min time lags. Should be overriden in child class for problems with min time lags.

Returns:: list of task1, task2, offset meaning start(task1) + offset <= start(task2)

update_time_lags() → None[source]

Method to call when time lags have been updated.

Clear cache from consolidated precedence constraints.

Returns:

class discrete_optimization.generic_tasks_tools.timelag.TimelagSolution(problem: Problem)[source]

Bases: SchedulingSolution[Task], Generic[Task]

Class for solution of problems having time lags between tasks.

check_time_lags() → bool[source]: check whether time lags are respected.

problem: TimelagProblem[Task]

discrete_optimization.generic_tasks_tools.timelag.consolidate_max_time_lags(timelags: list[tuple[Task, Task, int]]) → list[tuple[Task, Task, int]][source]

Get consolidated min time lags.

It merges min time lags targeting same tasks, taking the most restrictive offset.

Args:
timelags:

Returns:

discrete_optimization.generic_tasks_tools.timelag.consolidate_min_time_lags(timelags: list[tuple[Task, Task, int]]) → list[tuple[Task, Task, int]][source]

Get consolidated min time lags.

It merges min time lags targeting same tasks, taking the most restrictive offset.

Args:
timelags:

Returns:

discrete_optimization.generic_tasks_tools.timewindow module

class discrete_optimization.generic_tasks_tools.timewindow.TimewindowProblem[source]

Bases: SchedulingProblem[Task], Generic[Task]

Class for problem having time windows between tasks.

get_makespan_lower_bound() → int[source]

Get a lower bound on global makespan.

Time windows on last tasks can be used to get a better makespan lower bound.

get_task_bound(task: Task, start_or_end: StartOrEnd, min_or_max: MinOrMax) → int[source]

Get a lower or upper bound on a task start or end.

Parameters:

task
start_or_end
min_or_max – min -> lower bound, max -> upper bound

Returns:

get_task_start_or_end_lower_bound(task: Task, 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: Task, 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:

class discrete_optimization.generic_tasks_tools.timewindow.TimewindowSolution(problem: Problem)[source]

Bases: SchedulingSolution[Task], Generic[Task]

Class for solution of problems having time windows between tasks.

check_time_windows() → bool[source]: check whether time windows are respected.

problem: TimewindowProblem[Task]

discrete_optimization.generic_tasks_tools.utils module

discrete_optimization.generic_tasks_tools.utils.get_mandatory_methods_to_implement(cls: type[TasksProblem]) → dict[str, Callable[[...], Any]][source]

Find methods mandatory to implement when deriving from the given class.

Parameters:: cls – class to inspect
Returns:: A dictionary mapping abstract method names to their class methods.

Example

>>> from discrete_optimization.generic_tasks_tools.generic_scheduling import GenericSchedulingProblem
>>> get_mandatory_methods_to_implement(GenericSchedulingProblem)
{...}

discrete_optimization.generic_tasks_tools.utils.get_optional_methods_to_override(cls: type[TasksProblem] | TasksProblem, include_subclass_overrides: bool = False) → dict[str, Callable[[...], Any]][source]

Find methods likely to be overriden when deriving from the given task problem class.

It finds methods decorated with @optional_override. If the option is activated, it also finds such methods already overriden (including abstract methods).

Parameters:

cls – The class (or instance) to inspect.
include_subclass_overrides – If True, also includes - methods that overrode an @optional_override method from a parent mixin without re-applying the decorator - methods that overrode an @abstractmethod method from a parent mixin

Returns:

A dictionary mapping method names to their class methods.

Example

>>> from discrete_optimization.generic_tasks_tools.generic_scheduling import GenericSchedulingProblem
>>> optional_methods = get_optional_methods_to_override(GenericSchedulingProblem)
>>> optional_methods_including_subclasses = get_optional_methods_to_override(GenericSchedulingProblem, include_subclass_overrides=True)
>>> assert len(optional_methods_including_subclasses) > len(optional_methods)

discrete_optimization.generic_tasks_tools.utils.optional_override(funcobj)[source]

A decorator indicating primitive methods likely to be overriden when creating a new task problem class.

Many tasks mixins are defining abstract methods to be implemented. To reduce the burden when implementing a new tasks problem class, some other primitive methods have a default implementation. Other methods of the mixins are then derived from these primitive methods.

This decorator helps the problem developper to identify such primitive methods with default implementation that he could be interested in overriding.

Note: This does not modify the function behaviour, and is merely used as documentation.

discrete_optimization.generic_tasks_tools package

Subpackages

Submodules

discrete_optimization.generic_tasks_tools.allocation module

discrete_optimization.generic_tasks_tools.alternative_subproblems module

discrete_optimization.generic_tasks_tools.base module

discrete_optimization.generic_tasks_tools.calendar_resource module

discrete_optimization.generic_tasks_tools.cumulative_resource module

discrete_optimization.generic_tasks_tools.entities module

discrete_optimization.generic_tasks_tools.enums module

discrete_optimization.generic_tasks_tools.generic_scheduling module

discrete_optimization.generic_tasks_tools.generic_scheduling_impl module

discrete_optimization.generic_tasks_tools.generic_scheduling_utils module

discrete_optimization.generic_tasks_tools.multimode module

discrete_optimization.generic_tasks_tools.multimode_scheduling module

discrete_optimization.generic_tasks_tools.no_overlap module

discrete_optimization.generic_tasks_tools.non_renewable_resource module

discrete_optimization.generic_tasks_tools.plot_utils module

discrete_optimization.generic_tasks_tools.precedence module

discrete_optimization.generic_tasks_tools.precedence_scheduling module

discrete_optimization.generic_tasks_tools.resource_blocking module

discrete_optimization.generic_tasks_tools.scheduling module

discrete_optimization.generic_tasks_tools.skill module

discrete_optimization.generic_tasks_tools.solvers_map module

discrete_optimization.generic_tasks_tools.timelag module

discrete_optimization.generic_tasks_tools.timewindow module

discrete_optimization.generic_tasks_tools.utils module

Module contents