Adding a Fixed Duration Interruptible Task

processscheduler/__init__.py

@@ -56,6 +56,7 @@
     ZeroDurationTask,
     FixedDurationTask,
     VariableDurationTask,
+    FixedDurationInterruptibleTask,
 )
 from processscheduler.constraint import *
 from processscheduler.task_constraint import *

processscheduler/resource_constraint.py

@@ -25,7 +25,10 @@
 from processscheduler.resource import Worker, CumulativeWorker, SelectWorkers
 from processscheduler.constraint import ResourceConstraint
 from processscheduler.util import sort_no_duplicates
-from processscheduler.task import VariableDurationTask
+from processscheduler.task import (
+    VariableDurationTask,
+    FixedDurationInterruptibleTask,
+)
 
 
 class WorkLoad(ResourceConstraint):
@@ -242,7 +245,7 @@ def __init__(self, **data):
         if isinstance(self.resource, Worker):
             workers = [self.resource]
         elif isinstance(self.resource, CumulativeWorker):
-            workers = self.resource.cumulative_workers
+            workers = self.resource._cumulative_workers
 
         resource_assigned = False
 
@@ -315,6 +318,9 @@ def __init__(self, **data) -> None:
                 overlaps = []
 
                 is_interruptible = isinstance(task, VariableDurationTask)
+                is_fixed_interruptible = isinstance(
+                    task, FixedDurationInterruptibleTask
+                )
 
                 for (
                     interval_lower_bound,
@@ -335,7 +341,6 @@ def __init__(self, **data) -> None:
 
                     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(
@@ -365,6 +370,9 @@ def __init__(self, **data) -> None:
                         conds.append(
                             task._duration <= task.max_duration + total_overlap
                         )
+                elif is_fixed_interruptible:
+                    total_overlap = z3.Sum(*overlaps)
+                    conds.append(task._overlap == total_overlap)
 
             # TODO: remove AND, as the solver does that anyways?
             self.set_z3_assertions(z3.And(*conds))
@@ -421,7 +429,7 @@ def __init__(self, **data) -> None:
         if isinstance(self.resource, Worker):
             workers = [self.resource]
         elif isinstance(self.resource, CumulativeWorker):
-            workers = self.resource.cumulative_workers
+            workers = self.resource._cumulative_workers
 
         resource_assigned = False
 
@@ -433,6 +441,9 @@ def __init__(self, **data) -> None:
 
                 # check if the task allows variable duration
                 is_interruptible = isinstance(task, VariableDurationTask)
+                is_fixed_interruptible = isinstance(
+                    task, FixedDurationInterruptibleTask
+                )
 
                 duration = end_task_i - start_task_i
                 folded_start_task_i = (start_task_i - self.offset) % self.period
@@ -488,9 +499,8 @@ def __init__(self, **data) -> None:
                     )
                     overlaps.append(overlap)
 
-                    if is_interruptible:
+                    if is_interruptible or is_fixed_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(
@@ -520,6 +530,9 @@ def __init__(self, **data) -> None:
                         conds.append(
                             task._duration <= task.max_duration + total_overlap
                         )
+                elif is_fixed_interruptible:
+                    total_overlap = z3.Sum(*overlaps)
+                    conds.append(task._overlap == total_overlap)
 
             # TODO: add AND only of mask is set?
             core = z3.And(*conds)

processscheduler/task.py

@@ -74,7 +74,9 @@ def __init__(self, **data) -> None:
         super().__init__(**data)
 
         # workers required to process the task
-        self._required_resources = []  # type: List[Union[Worker, CumulativeWorker, SelectWorkers]]
+        self._required_resources = (
+            []
+        )  # type: List[Union[Worker, CumulativeWorker, SelectWorkers]]
 
         # z3 Int variables
         self._start = z3.Int(f"{self.name}_start")  # type: z3.ArithRef
@@ -88,7 +90,9 @@ def __init__(self, **data) -> None:
             raise AssertionError("No active problem. First create a SchedulingProblem")
         # the task_number is an integer that is incremented each time
         # a task is created. The first task has number 1, the second number 2 etc.
-        self._task_number = processscheduler.base.active_problem.add_task(self)  # type: int
+        self._task_number = processscheduler.base.active_problem.add_task(
+            self
+        )  # type: int
 
         # the release date
         if self.release_date is not None:
@@ -268,6 +272,21 @@ def __init__(self, **data) -> None:
         self.set_assertions(assertions)
 
 
+class FixedDurationInterruptibleTask(Task):
+    duration: PositiveInt
+
+    def __init__(self, **data) -> None:
+        super().__init__(**data)
+        self._overlap = z3.Int(f"{self.name}_overlap")
+        assertions = [
+            self._end - self._start == self.duration + self._overlap,
+            self._start >= 0,
+            self._overlap >= 0,
+        ]
+
+        self.set_assertions(assertions)
+
+
 class VariableDurationTask(Task):
     """The duration can take any value, computed by the solver."""