move_operation method added

b_asic/schema.py

@@ -14,7 +14,6 @@ import io
 
 from b_asic.signal_flow_graph import SFG
 from b_asic.graph_component import GraphID
-from b_asic.operation import Operation
 from b_asic.special_operations import Delay, Output
 
 
@@ -108,6 +107,22 @@ class Schema:
     def schedule_time(self) -> int:
         return self._schedule_time
 
+    def increase_time_resolution(self, factor: int) -> None:
+        raise NotImplementedError
+
+    def decrease_time_resolution(self, factor: int) -> None:
+        raise NotImplementedError
+
+    def move_operation(self, op_id: GraphID, time: int) -> None:
+        assert op_id in self._start_times, "No operation with the specified op_id in this schema."
+        if time < 0:
+            if -time > self.backward_slack(op_id):
+                raise ValueError
+        else:
+            if time > self.forward_slack(op_id):
+                raise ValueError
+        self._start_times[op_id] += time
+
     def _remove_delays(self) -> None:
         delay_list = self._sfg.find_by_type_name(Delay.type_name())
         while delay_list:
@@ -270,6 +285,7 @@ class Schema:
         plt.show()
 
     def _repr_svg_(self):
+        plt.figure()
         self._plot_schedule()
         f = io.StringIO()
         plt.savefig(f, format='svg')

test/test_schema.py

 """
 B-ASIC test suite for the schema module and Schema class.
 """
+import pytest
 
 from b_asic import Schema, Addition, ConstantMultiplication
 
@@ -77,3 +78,58 @@ class TestInit:
 
         assert schema.forward_slack(precedence_sfg_delays.find_by_name("A2")[0].graph_id) == 0
         assert schema.backward_slack(precedence_sfg_delays.find_by_name("A2")[0].graph_id) == 16
+
+
+class TestRescheduling:
+    def test_move_operation(self, precedence_sfg_delays):
+        precedence_sfg_delays.set_latency_of_type(Addition.type_name(), 4)
+        precedence_sfg_delays.set_latency_of_type(ConstantMultiplication.type_name(), 3)
+
+        schema = Schema(precedence_sfg_delays, scheduling_alg="ASAP")
+
+        schema.move_operation(precedence_sfg_delays.find_by_name("ADD3")[0].graph_id, 4)
+        schema.move_operation(precedence_sfg_delays.find_by_name("A2")[0].graph_id, 2)
+
+        start_times_names = dict()
+        for op_id, start_time in schema._start_times.items():
+            op_name = precedence_sfg_delays.find_by_id(op_id).name
+            start_times_names[op_name] = start_time
+
+        assert start_times_names == {"IN1": 0, "C0": 0, "B1": 0, "B2": 0, "ADD2": 3, "ADD1": 7, "Q1": 11,
+                                     "A0": 14, "A1": 0, "A2": 2, "ADD3": 7, "ADD4": 17, "OUT1": 21}
+
+    def test_move_operation_slack_after_rescheduling(self, precedence_sfg_delays):
+        precedence_sfg_delays.set_latency_of_type(Addition.type_name(), 1)
+        precedence_sfg_delays.set_latency_of_type(ConstantMultiplication.type_name(), 3)
+
+        schema = Schema(precedence_sfg_delays, scheduling_alg="ASAP")
+        schema.move_operation(precedence_sfg_delays.find_by_name("ADD3")[0].graph_id, 4)
+        assert schema.forward_slack(precedence_sfg_delays.find_by_name("ADD3")[0].graph_id) == 3
+        assert schema.backward_slack(precedence_sfg_delays.find_by_name("ADD3")[0].graph_id) == 4
+
+        assert schema.forward_slack(precedence_sfg_delays.find_by_name("A2")[0].graph_id) == 4
+        assert schema.backward_slack(precedence_sfg_delays.find_by_name("A2")[0].graph_id) == 16
+
+        schema.move_operation(precedence_sfg_delays.find_by_name("A2")[0].graph_id, 2)
+
+        assert schema.backward_slack(precedence_sfg_delays.find_by_name("ADD3")[0].graph_id) == 3
+        assert schema.backward_slack(precedence_sfg_delays.find_by_name("ADD3")[0].graph_id) == 2
+
+        assert schema.forward_slack(precedence_sfg_delays.find_by_name("A2")[0].graph_id) == 2
+        assert schema.backward_slack(precedence_sfg_delays.find_by_name("A2")[0].graph_id) == 18
+
+    def test_move_operation_incorrect_move_backward(self, precedence_sfg_delays):
+        precedence_sfg_delays.set_latency_of_type(Addition.type_name(), 1)
+        precedence_sfg_delays.set_latency_of_type(ConstantMultiplication.type_name(), 3)
+
+        schema = Schema(precedence_sfg_delays, scheduling_alg="ASAP")
+        with pytest.raises(ValueError):
+            schema.move_operation(precedence_sfg_delays.find_by_name("ADD3")[0].graph_id, -4)
+
+    def test_move_operation_incorrect_move_forward(self, precedence_sfg_delays):
+        precedence_sfg_delays.set_latency_of_type(Addition.type_name(), 1)
+        precedence_sfg_delays.set_latency_of_type(ConstantMultiplication.type_name(), 3)
+
+        schema = Schema(precedence_sfg_delays, scheduling_alg="ASAP")
+        with pytest.raises(ValueError):
+            schema.move_operation(precedence_sfg_delays.find_by_name("ADD3")[0].graph_id, 10)