from __future__ import annotations
import logging
import numpy as np
from discrete_optimization.generic_tasks_tools.scheduling import (
SchedulingProblem,
SchedulingSolution,
Task,
)
from discrete_optimization.generic_tasks_tools.utils import optional_override
logger = logging.getLogger(__name__)
[docs]
class NoOverlapProblem(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.
"""
[docs]
@optional_override
def get_no_overlap(self) -> set[frozenset[Task]]:
"""
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.
"""
return set()
[docs]
def get_forbidden_intervals(self, task: Task) -> list[tuple[int, int]]:
"""Get fixed intervals that should not overlap with given task.
Default to empty list. To be overridden in child classes.
Args:
task:
Returns:
List of intervals (start, end) with start <= end (will not be checked)
"""
return []
[docs]
def is_task_in_nooverlap_constraint(self, task: Task) -> bool:
"""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)`.
"""
return len(self.get_forbidden_intervals(task)) > 0 or any(
task in no_overlapping_tasks
for no_overlapping_tasks in self.get_no_overlap()
)
[docs]
class NoOverlapSolution(SchedulingSolution[Task]):
"""Solution for problem with precedence constraints."""
problem: NoOverlapProblem[Task]
[docs]
def check_no_overlap(self) -> bool:
# Doesnt work perfectly for zero duration tasks,
# this is more equivalent to cumulative with capacity 1.
for tasks in self.problem.get_no_overlap():
# remove absent optional tasks
tasks = frozenset(task for task in tasks if self.is_present(task))
min_start = min(self.get_start_time(t) for t in tasks)
max_end = max(self.get_end_time(t) for t in tasks)
cumul_use = np.zeros((max_end - min_start))
for task in tasks:
st, end = self.get_start_time(task), self.get_end_time(task)
if end == st:
continue
if np.max(cumul_use[(st - min_start) : (end - min_start)]) == 1:
logger.debug(f"Task {task} has overlap inside {tasks}")
return False
cumul_use[(st - min_start) : (end - min_start)] += 1
return True
[docs]
def check_forbidden_intervals(self) -> bool:
for task in self.get_present_tasks():
intervals = self.problem.get_forbidden_intervals(task)
if len(intervals) > 0:
start1 = self.get_start_time(task)
end1 = self.get_end_time(task)
if any(
_check_intervals_intersect(start1, end1, start2, end2)
for start2, end2 in intervals
):
logger.debug(f"Task {task} overlaps with its forbidden intervals.")
return False
return True
def _check_intervals_intersect(start1: int, end1: int, start2: int, end2: int) -> bool:
"""Check whether two intervals are intersecting
We assume that start1<end1 and start2<end2.
Args:
start1: start of first interval
end1: end of first interval
start2: start of second interval
end2: end of second interval
Returns:
"""
return not ((end1 <= start2) or (end2 <= start1))