discrete_optimization.rcpsp_resource_dependent package

Subpackages

• discrete_optimization.rcpsp_resource_dependent.solvers package
  • Submodules
  • discrete_optimization.rcpsp_resource_dependent.solvers.cpsat module
    • CpSatRcpspResourceDependentSolver
      • CpSatRcpspResourceDependentSolver.convert_task_variables_to_solution()
      • CpSatRcpspResourceDependentSolver.problem
  • Module contents

Submodules

discrete_optimization.rcpsp_resource_dependent.generator module

Utilities for generating resource-dependent RCPSP problems from standard RCPSP problems.

This module provides functions to transform regular RCPSP problems into resource-dependent variants by adding mode-dependent resource consumption patterns.

class discrete_optimization.rcpsp_resource_dependent.generator.DependencyStrategy(*values)

Bases: Enum

Strategy for adding resource dependencies.

MIXED = 'mixed'

PREDECESSOR_BASED = 'predecessor_based'

RANDOM = 'random'

RESOURCE_CONTENTION = 'resource_contention'

discrete_optimization.rcpsp_resource_dependent.generator.add_simple_resource_dependency(base_problem: RcpspProblem, task: Hashable, resource: str, dependent_task: Hashable, mode_to_value: dict[int, int]) → RcpspResourceDependentProblem

Add a simple resource dependency to a specific task-resource pair.

This is a low-level utility for manually creating specific dependencies.

Parameters:

• base_problem – Base RCPSP problem

• task – Task whose resource consumption should depend on another task

• resource – Resource name

• dependent_task – Task whose mode affects the consumption

• mode_to_value – Mapping from dependent_task’s modes to resource consumption values

Returns:

New RcpspResourceDependentProblem with the dependency added

Example

>>> # Make task "2" resource "R1" consumption depend on task "1" mode
>>> rd_problem = add_simple_resource_dependency(
...     base_problem=rcpsp_problem,
...     task="2",
...     resource="R1",
...     dependent_task="1",
...     mode_to_value={1: 2, 2: 5}  # mode 1 -> 2 units, mode 2 -> 5 units
... )

discrete_optimization.rcpsp_resource_dependent.generator.generate_resource_dependent_problem(base_problem: RcpspProblem, dependency_strategy: DependencyStrategy = DependencyStrategy.PREDECESSOR_BASED, dependency_probability: float = 0.3, variation_factor: float = 0.5, seed: int | None = None) → RcpspResourceDependentProblem

Generate a resource-dependent RCPSP problem from a standard RCPSP problem.

Parameters:

• base_problem – Base RCPSP problem to transform

• dependency_strategy – Strategy for selecting which tasks have dependencies

• dependency_probability – Probability that a task-resource pair becomes dependent (0.0-1.0)

• variation_factor – Factor controlling variation in resource consumption (0.0-1.0). Higher values mean more variation between different mode configurations.

• seed – Random seed for reproducibility

Returns:

RcpspResourceDependentProblem with added resource dependencies

Example

>>> from discrete_optimization.rcpsp.parser import get_data_available, parse_file
>>> from discrete_optimization.rcpsp_resource_dependent.generator import (
...     generate_resource_dependent_problem,
...     DependencyStrategy
... )
>>> # Load a standard RCPSP instance
>>> files = get_data_available()
>>> rcpsp_problem = parse_file(files[0])
>>> # Generate resource-dependent variant
>>> rd_problem = generate_resource_dependent_problem(
...     base_problem=rcpsp_problem,
...     dependency_strategy=DependencyStrategy.PREDECESSOR_BASED,
...     dependency_probability=0.3,
...     variation_factor=0.5,
...     seed=42
... )

discrete_optimization.rcpsp_resource_dependent.generator.validate_resource_dependent_problem(problem: RcpspResourceDependentProblem) → dict[str, Any]

Validate a resource-dependent problem and return statistics.

Parameters:

problem – Problem to validate

Returns:

• num_tasks: Total number of tasks

• num_dependent_consumptions: Number of dependent resource consumptions

• num_fixed_consumptions: Number of fixed resource consumptions

• dependency_ratio: Ratio of dependent to total consumptions

• tasks_with_dependencies: Set of tasks having at least one dependency

Return type:

Dictionary with validation statistics including

Example

>>> stats = validate_resource_dependent_problem(rd_problem)
>>> print(f"Dependency ratio: {stats['dependency_ratio']:.2%}")

discrete_optimization.rcpsp_resource_dependent.problem module

class discrete_optimization.rcpsp_resource_dependent.problem.RcpspResourceDependentProblem(resources: dict[str, int | list[int]], non_renewable_resources: list[str], mode_details: dict[Hashable, dict[int, dict[str, int]]], successors: dict[Hashable, list[Hashable]], horizon: int, tasks_list: list[Hashable] | None = None, source_task: Hashable | None = None, sink_task: Hashable | None = None)

Bases: GenericSchedulingProblem[Task, None, None, NonSkillCumulativeResource, NonRenewableResource], WithoutSkillProblem[Task, None, NonSkillCumulativeResource, None], WithoutAllocationProblem[Task]

RCPSP with resource-dependent consumption.

Supports consumption that depends on other tasks’ modes.

Encoding in mode_details:

Fixed consumption (int):

>>> mode_details = {
...     "task_A": {0: {"duration": 5, "electricity": 10}}
... }

Dependent consumption (dict mapping):

>>> mode_details = {
...     "task_A": {
...         0: {
...             "duration": 5,
...             "electricity": {
...                 frozenset([("task_B", 0)]): 10,  # B in mode 0: use 10
...                 frozenset([("task_B", 1)]): 8,   # B in mode 1: use 8
...             }
...         }
...     }
... }

See CumulativeResourceProblem for full documentation.

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

Evaluate a given solution object for the given problem.

This method should return a dictionnary of KPI, that can be then used for mono or multiobjective optimization.

Parameters:

variable (Solution) – the Solution object to evaluate.

Returns: dictionnary of float kpi for the solution.

get_cumulative_resource_consumption(resource: CumulativeResource, task: Task, mode: int) → int

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]

Get resource consumption mapping from mode_details.

Returns the dict for dependent tasks, or {frozenset([]): value} for standard tasks.

get_makespan_upper_bound() → int

Get an upper bound on global makespan.

get_non_renewable_resource_capacity(resource: NonRenewableResource) → int

Get resource max capacity

Parameters:

resource

Returns:

get_non_renewable_resource_consumption(resource: NonRenewableResource, task: Task, mode: int) → int

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]

get_objective_register() → ObjectiveRegister

Returns the objective definition.

Returns (ObjectiveRegister): object defining the objective criteria.

get_precedence_constraints() → dict[Task, Iterable[Task]]

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

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

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_solution_type() → type[Solution]

Returns the class implementation of a Solution.

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

get_task_mode_duration(task: Task, mode: int) → int

Get task duration according to mode.

Parameters:

• task

• mode – not used for single-mode problems

Returns:

get_task_modes(task: Task) → set[int]

Retrieve mode found for given task.

Parameters:

task

Returns:

is_cumulative_resource_task_mode_consumption_dependent(resource: CumulativeResource, task: Task, mode: int) → bool

Check if consumption depends on other tasks’ modes.

Determined by type in mode_details: int → False, dict → True, missing → None.

is_non_renewable_resource_task_mode_consumption_dependent(resource: NonRenewableResource, task: Task, mode: int)

property non_renewable_resources_list: list[NonRenewableResource]

Non-renewable resources used by the tasks.

property non_skill_cumulative_resources_list: list[Skill]

List of cumulative resources that are not skills.

property tasks_list: list[Task]

List of all tasks to schedule or allocate to.

class discrete_optimization.rcpsp_resource_dependent.problem.RcpspResourceDependentSolution(problem: RcpspResourceDependentProblem, schedule: dict[Task, tuple[int, int]], modes: dict[Task, int])

Bases: GenericSchedulingSolution[Task, None, None, NonSkillCumulativeResource, NonRenewableResource], WithoutSkillSolution[Task, None, NonSkillCumulativeResource, None], WithoutAllocationSolution[Task]

copy() → Solution

Deep copy of the solution.

The copy() function should return a new object containing the same input as the current object, that respects the following expected behaviour: -y = x.copy() -if do some inplace change of y, the changes are not done in x.

Returns: a new object from which you can manipulate attributes without changing the original object.

get_end_time(task: Task) → int

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: Task) → int

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: Task) → int

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:

Module contents