Merge pull request #142 from jbdyn/141-periodic-constraints-interruptible-tasks

Resolve "Define time spans periodically for ResourceUnavailable"

Author: tpaviot

processscheduler/__init__.py

@@ -63,9 +63,12 @@
     SameWorkers,
     DistinctWorkers,
     ResourceUnavailable,
+    ResourcePeriodicallyUnavailable,
     WorkLoad,
     ResourceTasksDistance,
     ResourceNonDelay,
+    ResourceInterrupted,
+    ResourcePeriodicallyInterrupted,
 )
 from processscheduler.indicator_constraint import IndicatorTarget, IndicatorBounds
 from processscheduler.resource import Worker, CumulativeWorker, SelectWorkers

processscheduler/resource_constraint.py

@@ -25,6 +25,7 @@
 from processscheduler.resource import Worker, CumulativeWorker, SelectWorkers
 from processscheduler.constraint import ResourceConstraint
 from processscheduler.util import sort_no_duplicates
+from processscheduler.task import VariableDurationTask
 
 
 class WorkLoad(ResourceConstraint):
@@ -197,6 +198,333 @@ def __init__(self, **data) -> None:
             )
 
 
+class ResourcePeriodicallyUnavailable(ResourceConstraint):
+    """
+    This constraint sets the unavailability of a resource in terms of time intervals in one period.
+
+    Parameters:
+    - resource: The resource for which unavailability is defined.
+    - list_of_time_intervals: A list of time intervals *in one period* during which the resource is unavailable for any task.
+      For example, [(0, 2), (5, 8)] represents time intervals from 0 to 2 and from 5 to 8 in one period.
+    - period: The length of one period after which to repeat the list of time intervals.
+      For example, setting this to 5 with [(2, 4)] gives unavailabilities at (2, 4), (7, 9), (12, 14), ...
+    - start: The start after which repeating the list of time intervals is active (default is 0).
+    - offset: The shift of the repeated list of time intervals (default is 0).
+      It might be desired to set also the start parameter to the same value, as otherwise the pattern shifts in from the left or the right into the schedule.
+    - end: The end until which repeating the list of time intervals is activate (default is None).
+    - optional (bool, optional): Whether the constraint is optional (default is False).
+    """
+
+    resource: Union[Worker, CumulativeWorker]
+    list_of_time_intervals: List[Tuple[int, int]]
+    period: int
+    start: int = 0
+    offset: int = 0
+    end: Union[int, None] = None
+
+
+    def __init__(self, **data):
+        """
+        Initialize a ResourceUnavailable constraint.
+
+        :param resource: The resource for which unavailability is defined.
+        :param list_of_time_intervals: A list of time intervals *in one period* during which the resource is unavailable for any task.
+          For example, [(0, 2), (5, 8)] represents time intervals from 0 to 2 and from 5 to 8.
+        :param period: The length of one period after which to repeat the list of time intervals.
+          For example, setting this to 5 with [(2, 4)] gives unavailabilities at (2, 4), (7, 9), (12, 14), ...
+        :param start: The start after which repeating the list of time intervals is active (default is 0).
+        :param offset: The shift of the repeated list of time intervals (default is 0).
+          It might be desired to set also the start parameter to the same value, as otherwise the pattern shifts in from the left or the right into the schedule.
+        :param end: The end until which repeating the list of time intervals is activate (default is None).
+        :param optional: Whether the constraint is optional (default is False).
+        """
+        super().__init__(**data)
+
+        if isinstance(self.resource, Worker):
+            workers = [self.resource]
+        elif isinstance(self.resource, CumulativeWorker):
+            workers = self.resource.cumulative_workers
+
+        resource_assigned = False
+
+        for interval_lower_bound, interval_upper_bound in self.list_of_time_intervals:
+            for worker in workers:
+                for start_task_i, end_task_i in worker.get_busy_intervals():
+                    resource_assigned = True
+                    duration = end_task_i - start_task_i
+                    conds = [
+                        z3.Xor(
+                            (start_task_i - self.offset) % self.period >= interval_upper_bound,
+                            (start_task_i - self.offset) % self.period + duration <= interval_lower_bound
+                        )
+                    ]
+
+                    if self.start > 0:
+                        conds.append(end_task_i <= self.start)
+                    if self.end is not None:
+                        conds.append(start_task_i >= self.end)
+
+                    if len(conds) > 1:
+                        self.set_z3_assertions(z3.Or(*conds))
+                    else:
+                        self.set_z3_assertions(*conds)
+
+        if not resource_assigned:
+            raise AssertionError(
+                "The resource is not assigned to any task. Please first assign the resource to one or more tasks, and then add the ResourcePeriodicallyUnavailable constraint."
+            )
+
+
+class ResourceInterrupted(ResourceConstraint):
+    """
+    This constraint sets the interrupts of a resource in terms of time intervals.
+
+    Parameters:
+    - resource: The resource for which interrupts are defined.
+    - list_of_time_intervals: A list of time intervals during which the resource is interrupting any task.
+      For example, [(0, 2), (5, 8)] represents time intervals from 0 to 2 and from 5 to 8.
+    - optional (bool, optional): Whether the constraint is optional (default is False).
+    """
+
+    resource: Union[Worker, CumulativeWorker]
+    list_of_time_intervals: List[Tuple[int, int]]
+
+    def __init__(self, **data) -> None:
+        """
+        Initialize a ResourceInterrupted constraint.
+
+        :param resource: The resource for which interrupts are defined.
+        :param list_of_time_intervals: A list of time intervals during which the resource is interrupting any task.
+          For example, [(0, 2), (5, 8)] represents time intervals from 0 to 2 and from 5 to 8.
+        :param optional: Whether the constraint is optional (default is False).
+        """
+        super().__init__(**data)
+
+        if isinstance(self.resource, Worker):
+            workers = [self.resource]
+        elif isinstance(self.resource, CumulativeWorker):
+            workers = self.resource.cumulative_workers
+
+        resource_assigned = False
+
+        for worker in workers:
+            conds = []
+            for task, (start_task_i, end_task_i) in worker._busy_intervals.items():
+                resource_assigned = True
+                overlaps = []
+
+                is_interruptible = isinstance(task, VariableDurationTask)
+
+                for interval_lower_bound, interval_upper_bound in self.list_of_time_intervals:
+                    overlap_condition = z3.Not(z3.Xor(
+                        start_task_i >= interval_upper_bound,
+                        end_task_i <= interval_lower_bound
+                    ))
+                    overlap = z3.If(
+                        overlap_condition,
+                        interval_upper_bound - interval_lower_bound,
+                        0
+                    )
+                    overlaps.append(overlap)
+
+                    if is_interruptible:
+                        # just make sure that the task does not start or end within the time interval...
+                        # TODO: account for zero-duration?
+                        conds.extend([
+                            z3.Xor(
+                                start_task_i <= interval_lower_bound,
+                                start_task_i >= interval_upper_bound
+                            ),
+                            z3.Xor(
+                                end_task_i <= interval_lower_bound,
+                                end_task_i >= interval_upper_bound
+                            )
+                        ])
+                    else:
+                        # ...otherwise make sure the task does not overlap with the time interval
+                        conds.append(
+                            z3.Xor(
+                                start_task_i >= interval_upper_bound,
+                                end_task_i <= interval_lower_bound
+                            )
+                        )
+
+                if is_interruptible:
+                    # add assertions for task duration based on the total count of overlapped periods
+                    total_overlap = z3.Sum(*overlaps)
+                    conds.append(
+                        task._duration >= task.min_duration + total_overlap
+                    )
+                    if task.max_duration is not None:
+                        conds.append(
+                            task._duration <= task.max_duration + total_overlap
+                        )
+
+            # TODO: remove AND, as the solver does that anyways?
+            self.set_z3_assertions(z3.And(*conds))
+
+        if not resource_assigned:
+            raise AssertionError(
+                "The resource is not assigned to any task. Please first assign the resource to one or more tasks, and then add the ResourceInterrupted constraint."
+            )
+
+
+class ResourcePeriodicallyInterrupted(ResourceConstraint):
+    """
+    This constraint sets the interrupts of a resource in terms of time intervals.
+
+    Parameters:
+    - resource: The resource for which interrupts are defined.
+    - list_of_time_intervals: A list of time intervals during which the resource is interrupting any task.
+      For example, [(0, 2), (5, 8)] represents time intervals from 0 to 2 and from 5 to 8.
+    - optional (bool, optional): Whether the constraint is optional (default is False).
+    - period: The length of one period after which to repeat the list of time intervals.
+      For example, setting this to 5 with [(2, 4)] gives unavailabilities at (2, 4), (7, 9), (12, 14), ...
+    - start: The start after which repeating the list of time intervals is active (default is 0).
+    - offset: The shift of the repeated list of time intervals (default is 0).
+      It might be desired to set also the start parameter to the same value, as otherwise the pattern shifts in from the left or the right into the schedule.
+    - end: The end until which repeating the list of time intervals is activate (default is None).
+    - optional (bool, optional): Whether the constraint is optional (default is False).
+    """
+
+    resource: Union[Worker, CumulativeWorker]
+    list_of_time_intervals: List[Tuple[int, int]]
+    period: int
+    start: int = 0
+    offset: int = 0
+    end: Union[int, None] = None
+
+    def __init__(self, **data) -> None:
+        """
+        Initialize a ResourceInterrupted constraint.
+
+        :param resource: The resource for which interrupts are defined.
+        :param list_of_time_intervals: A list of time intervals during which the resource is interrupting any task.
+          For example, [(0, 2), (5, 8)] represents time intervals from 0 to 2 and from 5 to 8.
+        :param optional: Whether the constraint is optional (default is False).
+        :param period: The length of one period after which to repeat the list of time intervals.
+          For example, setting this to 5 with [(2, 4)] gives unavailabilities at (2, 4), (7, 9), (12, 14), ...
+        :param start: The start after which repeating the list of time intervals is active (default is 0).
+        :param offset: The shift of the repeated list of time intervals (default is 0).
+          It might be desired to set also the start parameter to the same value, as otherwise the pattern shifts in from the left or the right into the schedule.
+        :param end: The end until which repeating the list of time intervals is activate (default is None).
+        :param optional: Whether the constraint is optional (default is False).
+        """
+        super().__init__(**data)
+
+        if isinstance(self.resource, Worker):
+            workers = [self.resource]
+        elif isinstance(self.resource, CumulativeWorker):
+            workers = self.resource.cumulative_workers
+
+        resource_assigned = False
+
+        for worker in workers:
+            conds = []
+            for task, (start_task_i, end_task_i) in worker._busy_intervals.items():
+                resource_assigned = True
+                overlaps = []
+
+                # check if the task allows variable duration
+                is_interruptible = isinstance(task, VariableDurationTask)
+
+                duration = end_task_i - start_task_i
+                folded_start_task_i = (start_task_i - self.offset) % self.period
+                folded_end_task_i = (end_task_i - self.offset) % self.period
+
+                for interval_lower_bound, interval_upper_bound in self.list_of_time_intervals:
+                    # intervals need to be defined in one period
+                    if interval_upper_bound > self.period:
+                        raise AssertionError(f"interval ({interval_lower_bound}, {interval_upper_bound}) exceeds period {self.period}")
+
+                    # if true, the folded task overlaps with the time interval in the first period
+                    crossing_condition = z3.Not(z3.Xor(
+                        # folded task is completely before the first time interval
+                        z3.And(
+                            folded_start_task_i <= interval_lower_bound,
+                            folded_start_task_i + duration % self.period <= interval_lower_bound,
+                        ),
+                        # folded task is completely between the first and second time interval
+                        z3.And(
+                            folded_start_task_i >= interval_upper_bound,
+                            folded_start_task_i + duration % self.period <= interval_lower_bound + self.period,
+                        )
+                    ))
+
+                    # if true, the task overlaps with at least one time interval
+                    overlap_condition = z3.Or(
+                        crossing_condition,
+                        # task does not fit between two intervals
+                        duration > interval_lower_bound + self.period - interval_upper_bound
+                    )
+
+                    # adjust the number of crossed time intervals
+                    crossings = z3.If(
+                        crossing_condition,
+                        duration / self.period + 1,
+                        duration / self.period
+                    )
+                    # calculate the total overlap for this particular time interval
+                    overlap = z3.If(
+                        overlap_condition,
+                        (interval_upper_bound - interval_lower_bound) * crossings,
+                        0
+                    )
+                    overlaps.append(overlap)
+
+                    if is_interruptible:
+                        # just make sure that the task does not start or end within one of the time intervals...
+                        # TODO: account for zero-duration?
+                        conds.extend([
+                            z3.Xor(
+                                folded_start_task_i <= interval_lower_bound,
+                                folded_start_task_i >= interval_upper_bound
+                            ),
+                            z3.Xor(
+                                folded_end_task_i <= interval_lower_bound,
+                                folded_end_task_i >= interval_upper_bound
+                            )
+                        ])
+                    else:
+                        # ...otherwise make sure the task does not overlap with any of time intervals
+                        conds.append(
+                            z3.Xor(
+                                folded_start_task_i >= interval_upper_bound,
+                                folded_start_task_i + duration <= interval_lower_bound
+                            )
+                        )
+
+                if is_interruptible:
+                    # add assertions for task duration based on the total count of overlapped periods
+                    total_overlap = z3.Sum(*overlaps)
+                    conds.append(
+                        task._duration >= task.min_duration + total_overlap
+                    )
+                    if task.max_duration is not None:
+                        conds.append(
+                            task._duration <= task.max_duration + total_overlap
+                        )
+
+            # TODO: add AND only of mask is set?
+            core = z3.And(*conds)
+
+            mask = [core]
+            if self.start > 0:
+                mask.append(end_task_i <= self.start)
+            if self.end is not None:
+                mask.append(start_task_i >= self.end)
+
+            if len(mask) > 1:
+                self.set_z3_assertions(z3.Or(*mask))
+            else:
+                self.set_z3_assertions(*mask)
+
+        if not resource_assigned:
+            raise AssertionError(
+                "The resource is not assigned to any task. Please first assign the resource to one or more tasks, and then add the ResourcePeriodicallyInterrupted constraint."
+            )
+
+
 class ResourceNonDelay(ResourceConstraint):
     """All tasks processed by the resource are contiguous, there's no idle while an operation
     if waiting for processing"""