from __future__ import annotations
import logging
from abc import abstractmethod
from collections.abc import Hashable, Iterable
from typing import Any, Generic, Optional, TypeVar
from discrete_optimization.generic_tasks_tools.base import (
Task,
TasksCpSolver,
TasksProblem,
TasksSolution,
)
from discrete_optimization.generic_tasks_tools.utils import optional_override
from discrete_optimization.generic_tools.cp_tools import SignEnum
logger = logging.getLogger(__name__)
UnaryResource = TypeVar("UnaryResource", bound=Hashable)
[docs]
class AllocationSolution(TasksSolution[Task], Generic[Task, UnaryResource]):
"""Class inherited by a solution for allocation problems."""
problem: AllocationProblem[Task, UnaryResource]
[docs]
@abstractmethod
def is_allocated(self, task: Task, unary_resource: UnaryResource) -> bool:
"""Return the usage of the unary resource for the given task.
Args:
task:
unary_resource:
Returns:
"""
...
[docs]
def is_present(self, task: Task) -> bool:
"""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.
"""
return len(self.get_task_allocation(task)) > 0
[docs]
def get_task_allocation(self, task: Task) -> set[UnaryResource]:
return {
unary_resource
for unary_resource in self.problem.unary_resources_list
if self.is_allocated(task=task, unary_resource=unary_resource)
}
[docs]
def get_default_tasks_n_unary_resources(
self,
tasks: Optional[Iterable[Task]] = None,
unary_resources: Optional[Iterable[UnaryResource]] = None,
) -> tuple[Iterable[Task], Iterable[UnaryResource]]:
return get_default_tasks_n_unary_resources(self.problem, tasks, unary_resources)
[docs]
def compute_nb_unary_resource_usages(
self,
unary_resources: Optional[Iterable[UnaryResource]] = None,
):
tasks, unary_resources = self.get_default_tasks_n_unary_resources(
self.get_present_tasks(), unary_resources
)
return sum(
self.is_allocated(task=task, unary_resource=unary_resource)
for task in tasks
for unary_resource in unary_resources
)
def _check_same_problem(self, ref: AllocationSolution[Task, UnaryResource]) -> None:
if self.problem is not ref.problem:
raise ValueError("We can compare only solutions for same problem.")
[docs]
def compute_nb_allocation_changes(
self,
ref: AllocationSolution[Task, UnaryResource],
tasks: Optional[Iterable[Task]] = None,
unary_resources: Optional[Iterable[UnaryResource]] = None,
) -> int:
self._check_same_problem(ref)
tasks, unary_resources = self.get_default_tasks_n_unary_resources(
tasks, unary_resources
)
return sum(
self.is_allocated(task=task, unary_resource=unary_resource)
!= ref.is_allocated(task=task, unary_resource=unary_resource)
for task in tasks
for unary_resource in unary_resources
)
[docs]
def compute_nb_tasks_allocated(self) -> int:
"""Compute number of tasks with at least a resource allocated."""
return sum(
any(
self.is_allocated(task=task, unary_resource=unary_resource)
for unary_resource in self.problem.unary_resources_list
)
for task in self.get_present_tasks()
)
[docs]
def compute_nb_unary_resources_used(self) -> int:
"""Compute number of unary resources allocated to at least one task."""
return sum(
any(
self.is_allocated(task=task, unary_resource=unary_resource)
for task in self.get_present_tasks()
)
for unary_resource in self.problem.unary_resources_list
)
[docs]
def check_same_allocation_as_ref(
self,
ref: AllocationSolution[Task, UnaryResource],
tasks: Optional[Iterable[Task]] = None,
unary_resources: Optional[Iterable[UnaryResource]] = None,
) -> bool:
self._check_same_problem(ref)
tasks, unary_resources = self.get_default_tasks_n_unary_resources(
tasks, unary_resources
)
return all(
self.is_allocated(task=task, unary_resource=unary_resource)
== ref.is_allocated(task=task, unary_resource=unary_resource)
for task in tasks
for unary_resource in unary_resources
)
[docs]
def check_allocation_consistency(self) -> bool:
for task in self.problem.tasks_list:
for unary_resource in self.problem.unary_resources_list:
if self.is_allocated(
task=task, unary_resource=unary_resource
) and not self.problem.is_compatible_task_unary_resource(
task=task, unary_resource=unary_resource
):
logger.debug(
f"Unary resource {unary_resource} is allocated to task {task} "
"but it is not compatible with it."
)
return False
for set_of_task in self.problem.get_same_unary_allocation():
set_unary = set(
[frozenset(self.get_task_allocation(t)) for t in set_of_task]
)
if len(set_unary) >= 2:
logger.debug(
f"Tasks {set_of_task} don't have the same unary resource allocation"
)
return False
return True
[docs]
class AllocationProblem(TasksProblem[Task], Generic[Task, UnaryResource]):
"""Base class for allocation problems.
An allocation problems consist in allocating resources to tasks.
"""
@property
@abstractmethod
def unary_resources_list(self) -> list[UnaryResource]:
"""Available unary resources.
It can correspond to employees (rcpsp-multiskill), teams (workforce-scheduling), or
a mix of several types.
"""
...
_map_unary_resource_to_index: Optional[dict[UnaryResource, int]] = None
[docs]
def get_unary_resource_from_index(self, i: int) -> UnaryResource:
return self.unary_resources_list[i]
[docs]
def get_index_from_unary_resource(self, unary_resource: UnaryResource) -> int:
if self._map_unary_resource_to_index is None:
self._map_unary_resource_to_index = {
unary_resource: i
for i, unary_resource in enumerate(self.unary_resources_list)
}
return self._map_unary_resource_to_index[unary_resource]
[docs]
@optional_override
def is_compatible_task_unary_resource(
self, task: Task, unary_resource: UnaryResource
) -> bool:
"""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.
"""
return True
[docs]
def compatible_unary_resources(self, task: Task) -> set[UnaryResource]:
return {
ur
for ur in self.unary_resources_list
if self.is_compatible_task_unary_resource(task, ur)
}
[docs]
@optional_override
def get_same_unary_allocation(self) -> list[set[Task]]:
"""
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:
"""
return []
[docs]
class AllocationCpSolver(TasksCpSolver[Task], Generic[Task, UnaryResource]):
"""Base class for solver managing constraints on allocation."""
problem: AllocationProblem[Task, UnaryResource]
[docs]
@abstractmethod
def add_constraint_on_task_unary_resource_allocation(
self, task: Task, unary_resource: UnaryResource, used: bool
) -> list[Any]:
"""Add constraint on allocation of given unary resource for the given task
Args:
task:
unary_resource:
used: if True, we enforce the allocation of `unary_resource` to `task`, else we prevent it
Returns:
resulting constraints
"""
...
[docs]
@abstractmethod
def add_constraint_on_nb_allocation_changes(
self, ref: AllocationSolution[Task, UnaryResource], nb_changes: int
) -> list[Any]:
"""Add contraint on maximal number of allocation changes from the given reference.
Args:
ref:
nb_changes: maximal number of changes
Returns:
resulting constraints
"""
...
[docs]
@abstractmethod
def add_constraint_on_total_nb_usages(
self, sign: SignEnum, target: int
) -> list[Any]: ...
[docs]
@abstractmethod
def add_constraint_on_unary_resource_nb_usages(
self, unary_resource: UnaryResource, sign: SignEnum, target: int
) -> list[Any]: ...
[docs]
def get_default_tasks_n_unary_resources(
self,
tasks: Optional[Iterable[Task]] = None,
unary_resources: Optional[Iterable[UnaryResource]] = None,
) -> tuple[Iterable[Task], Iterable[UnaryResource]]:
return get_default_tasks_n_unary_resources(self.problem, tasks, unary_resources)
[docs]
def add_constraint_same_allocation_as_ref(
self,
ref: AllocationSolution[Task, UnaryResource],
tasks: Optional[Iterable[Task]] = None,
unary_resources: Optional[Iterable[UnaryResource]] = None,
) -> list[Any]:
"""Add constraint to keep same allocation as the reference for the given tasks and unary resources subsets.
Args:
ref:
tasks:
unary_resources:
Returns:
resulting constraints
"""
constraints = []
tasks, unary_resources = self.get_default_tasks_n_unary_resources(
tasks, unary_resources
)
for unary_resource in unary_resources:
for task in tasks:
constraints += self.add_constraint_on_task_unary_resource_allocation(
task=task,
unary_resource=unary_resource,
used=ref.is_allocated(task=task, unary_resource=unary_resource),
)
return constraints
[docs]
@abstractmethod
def get_nb_tasks_allocated_variable(self) -> Any:
"""Construct and get the variable tracking number of tasks with at least a resource allocated.
Returns:
objective variable to minimize
"""
...
[docs]
@abstractmethod
def get_nb_unary_resources_used_variable(self) -> Any:
"""Construct and get the variable tracking number of tasks with at least a resource allocated.
Returns:
objective variable to minimize
"""
...
[docs]
def is_compatible_task_unary_resource(
self, task: Task, unary_resource: UnaryResource
) -> bool:
"""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.
"""
return self.problem.is_compatible_task_unary_resource(
task=task, unary_resource=unary_resource
)
[docs]
def get_default_tasks_n_unary_resources(
problem: AllocationProblem,
tasks: Optional[Iterable[Task]] = None,
unary_resources: Optional[Iterable[UnaryResource]] = None,
) -> tuple[Iterable[Task], Iterable[UnaryResource]]:
if tasks is None:
tasks = problem.tasks_list
if unary_resources is None:
unary_resources = problem.unary_resources_list
return tasks, unary_resources
NoUnaryResource = None
[docs]
class WithoutAllocationProblem(AllocationProblem[Task, NoUnaryResource], Generic[Task]):
"""Mixin to simplify deriving from GenericSchedulingProblem when no allocation is needed."""
@property
def unary_resources_list(self) -> list[NoUnaryResource]:
return []
[docs]
class WithoutAllocationSolution(
AllocationSolution[Task, NoUnaryResource], Generic[Task]
):
"""Mixin to simplify deriving from GenericSchedulingSolution when no allocation is needed."""
[docs]
def is_allocated(self, task: Task, unary_resource: UnaryResource) -> bool:
return False
[docs]
def check_allocation_consistency(self) -> bool:
return True