ruff format new periodic constraints and related tests see #142

Author: tpaviot

processscheduler/resource_constraint.py

@@ -222,7 +222,6 @@ class ResourcePeriodicallyUnavailable(ResourceConstraint):
     offset: int = 0
     end: Union[int, None] = None
 
-
     def __init__(self, **data):
         """
         Initialize a ResourceUnavailable constraint.
@@ -254,8 +253,10 @@ def __init__(self, **data):
                     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
+                            (start_task_i - self.offset) % self.period
+                            >= interval_upper_bound,
+                            (start_task_i - self.offset) % self.period + duration
+                            <= interval_lower_bound,
                         )
                     ]
 
@@ -315,46 +316,51 @@ def __init__(self, **data) -> None:
 
                 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
-                    ))
+                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
+                        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
-                            )
-                        ])
+                        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
+                                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
-                    )
+                    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
@@ -432,74 +438,84 @@ def __init__(self, **data) -> None:
                 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:
+                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}")
+                        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,
+                    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
+                        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
+                        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
+                        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
-                            )
-                        ])
+                        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
+                                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
-                    )
+                    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

test/test_resource_interrupted.py

@@ -36,11 +36,12 @@ def test_resource_interrupted_fixed_duration() -> None:
     assert solution.tasks[task_2.name].start == 8
     assert solution.tasks[task_2.name].end == 12
 
+
 def test_resource_interrupted_variable_duration() -> None:
     pb = ps.SchedulingProblem(name="variable_duration")
     task_1 = ps.VariableDurationTask(name="task1", min_duration=3)
     task_2 = ps.FixedDurationTask(name="task2", duration=4)
-    ps.TaskStartAt(task=task_1, value=0) # pin to have a more stable outcome
+    ps.TaskStartAt(task=task_1, value=0)  # pin to have a more stable outcome
     worker_1 = ps.Worker(name="Worker1")
     task_1.add_required_resource(worker_1)
     task_2.add_required_resource(worker_1)

test/test_resource_periodically_interrupted.py

@@ -26,9 +26,7 @@ def test_resource_periodically_interrupted_fixed_duration() -> None:
     task_1.add_required_resource(worker_1)
     task_2.add_required_resource(worker_1)
     ps.ResourcePeriodicallyInterrupted(
-        resource=worker_1,
-        list_of_time_intervals=[(1, 2), (4, 5)],
-        period=10
+        resource=worker_1, list_of_time_intervals=[(1, 2), (4, 5)], period=10
     )
 
     ps.ObjectiveMinimizeMakespan()
@@ -45,14 +43,12 @@ def test_resource_periodically_interrupted_variable_duration() -> None:
     pb = ps.SchedulingProblem(name="variable_duration")
     task_1 = ps.VariableDurationTask(name="task1", min_duration=3)
     task_2 = ps.FixedDurationTask(name="task2", duration=4)
-    ps.TaskStartAt(task=task_1, value=0) # pin to have a more stable outcome
+    ps.TaskStartAt(task=task_1, value=0)  # pin to have a more stable outcome
     worker_1 = ps.Worker(name="Worker1")
     task_1.add_required_resource(worker_1)
     task_2.add_required_resource(worker_1)
     ps.ResourcePeriodicallyInterrupted(
-        resource=worker_1,
-        list_of_time_intervals=[(1, 2), (4, 5)],
-        period=10
+        resource=worker_1, list_of_time_intervals=[(1, 2), (4, 5)], period=10
     )
 
     ps.ObjectiveMinimizeMakespan()
@@ -69,11 +65,15 @@ def test_resource_periodically_interrupted_assignment_assertion() -> None:
     ps.SchedulingProblem(name="assignment_assertion")
     worker_1 = ps.Worker(name="Worker1")
     with pytest.raises(AssertionError):
-        ps.ResourcePeriodicallyInterrupted(resource=worker_1, list_of_time_intervals=[(1, 3)], period=6)
+        ps.ResourcePeriodicallyInterrupted(
+            resource=worker_1, list_of_time_intervals=[(1, 3)], period=6
+        )
 
 
 def test_resource_periodically_interrupted_period_assertion() -> None:
     ps.SchedulingProblem(name="period_assertion")
     worker_1 = ps.Worker(name="Worker1")
     with pytest.raises(AssertionError):
-        ps.ResourcePeriodicallyInterrupted(resource=worker_1, list_of_time_intervals=[(1, 2), (3, 5)], period=4)
+        ps.ResourcePeriodicallyInterrupted(
+            resource=worker_1, list_of_time_intervals=[(1, 2), (3, 5)], period=4
+        )

test/test_resource_periodically_unavailable.py

@@ -24,7 +24,9 @@ def test_resource_periodically_unavailable_1() -> None:
     task_1 = ps.FixedDurationTask(name="task1", duration=3)
     worker_1 = ps.Worker(name="Worker1")
     task_1.add_required_resource(worker_1)
-    ps.ResourcePeriodicallyUnavailable(resource=worker_1, list_of_time_intervals=[(1, 3), (6, 8)], period=10)
+    ps.ResourcePeriodicallyUnavailable(
+        resource=worker_1, list_of_time_intervals=[(1, 3), (6, 8)], period=10
+    )
 
     solver = ps.SchedulingSolver(problem=pb)
     solution = solver.solve()
@@ -40,8 +42,12 @@ def test_resource_periodically_unavailable_2() -> None:
     task_1.add_required_resource(worker_1)
     # difference with the first one: build 2 constraints
     # merged using a and_
-    ps.ResourcePeriodicallyUnavailable(resource=worker_1, list_of_time_intervals=[(1, 3)], period=10)
-    ps.ResourcePeriodicallyUnavailable(resource=worker_1, list_of_time_intervals=[(6, 8)], period=10)
+    ps.ResourcePeriodicallyUnavailable(
+        resource=worker_1, list_of_time_intervals=[(1, 3)], period=10
+    )
+    ps.ResourcePeriodicallyUnavailable(
+        resource=worker_1, list_of_time_intervals=[(6, 8)], period=10
+    )
 
     # that should not change the problem solution
     solver = ps.SchedulingSolver(problem=pb)
@@ -59,8 +65,12 @@ def test_resource_periodically_unavailable_3() -> None:
     # difference with the previous ones: too much unavailability,
     # so possible solution
     # merged using a and_
-    ps.ResourcePeriodicallyUnavailable(resource=worker_1, list_of_time_intervals=[(1, 3)], period=10)
-    ps.ResourcePeriodicallyUnavailable(resource=worker_1, list_of_time_intervals=[(5, 8)], period=10)
+    ps.ResourcePeriodicallyUnavailable(
+        resource=worker_1, list_of_time_intervals=[(1, 3)], period=10
+    )
+    ps.ResourcePeriodicallyUnavailable(
+        resource=worker_1, list_of_time_intervals=[(5, 8)], period=10
+    )
 
     # that should not change the problem solution
     solver = ps.SchedulingSolver(problem=pb)
@@ -72,7 +82,9 @@ def test_resource_periodically_unavailable_4() -> None:
     ps.SchedulingProblem(name="ResourcePeriodicallyUnavailable4", horizon=10)
     worker_1 = ps.Worker(name="Worker1")
     with pytest.raises(AssertionError):
-        ps.ResourcePeriodicallyUnavailable(resource=worker_1, list_of_time_intervals=[(1, 3)], period=10)
+        ps.ResourcePeriodicallyUnavailable(
+            resource=worker_1, list_of_time_intervals=[(1, 3)], period=10
+        )
 
 
 def test_resource_periodically_unavailable_5() -> None:
@@ -86,9 +98,7 @@ def test_resource_periodically_unavailable_5() -> None:
     for task in (task_1, task_2, task_3):
         task.add_required_resource(worker_1)
     ps.ResourcePeriodicallyUnavailable(
-        resource=worker_1,
-        list_of_time_intervals=[(3, 5)],
-        period=5
+        resource=worker_1, list_of_time_intervals=[(3, 5)], period=5
     )
 
     solver = ps.SchedulingSolver(problem=pb)
@@ -117,29 +127,30 @@ def test_resource_periodically_unavailable_6() -> None:
     ps.ResourcePeriodicallyUnavailable(
         resource=worker_1,
         list_of_time_intervals=[(2, 4)],
-        offset=2, # shift interval to (4, 6)
+        offset=2,  # shift interval to (4, 6)
         start=3,  # end_task_i <= start, so it must be set to the task duration
-        end=14,   # unavailability interval at (14, 16), but it should be ignored
-        period=5
+        end=14,  # unavailability interval at (14, 16), but it should be ignored
+        period=5,
     )
     ps.ResourceUnavailable(
         resource=worker_1,
-        list_of_time_intervals=[(3, 4)] # leaving task_1 starting at 0 as only option
+        list_of_time_intervals=[(3, 4)],  # leaving task_1 starting at 0 as only option
     )
 
-    ps.ObjectiveMinimizeMakespan() # ensure dense packing
+    ps.ObjectiveMinimizeMakespan()  # ensure dense packing
     solver = ps.SchedulingSolver(problem=pb)
     solution = solver.solve()
     assert solution
     assert solution.tasks[task_1.name].start == 0  # unavailability at (0, 1) ignored
     assert solution.tasks[task_1.name].end == 3
     assert solution.tasks[task_2.name].start == 6
-    assert solution.tasks[task_2.name].end == 9    # expected unavailability
+    assert solution.tasks[task_2.name].end == 9  # expected unavailability
     assert solution.tasks[task_3.name].start == 11
     assert solution.tasks[task_3.name].end == 14
-    assert solution.tasks[task_4.name].start == 14 # unavailability at (14, 16) ignored
+    assert solution.tasks[task_4.name].start == 14  # unavailability at (14, 16) ignored
     assert solution.tasks[task_4.name].end == 17
 
+
 def test_resource_periodically_unavailable_7() -> None:
     pb = ps.SchedulingProblem(name="ResourcePeriodicallyUnavailable3")
     task_1 = ps.FixedDurationTask(name="task1", duration=2)
@@ -151,9 +162,7 @@ def test_resource_periodically_unavailable_7() -> None:
     for task in (task_1, task_2, task_3):
         task.add_required_resource(worker_1)
     ps.ResourcePeriodicallyUnavailable(
-        resource=worker_1,
-        list_of_time_intervals=[(1, 2), (3, 4)],
-        period=5
+        resource=worker_1, list_of_time_intervals=[(1, 2), (3, 4)], period=5
     )
 
     ps.ObjectiveMinimizeMakespan()