move_operation and slacks methods added

b_asic/schema.py

@@ -4,7 +4,7 @@ Contains the schema class for scheduling operations in an SFG.
 """
 
 from collections import defaultdict
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Tuple
 import matplotlib.pyplot as plt
 from matplotlib.ticker import MaxNLocator
 import numpy as np
@@ -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
 
 
@@ -97,6 +96,10 @@ class Schema:
                 slack = min(slack, usage_time - available_time)
         return slack
 
+    def slacks(self, op_id: GraphID) -> Tuple(int, int):
+        assert op_id in self._start_times, "No operation with the specified op_id in this schema."
+        return (self.backward_slack(op_id), self.forward_slack(op_id))
+
     def print_slacks(self) -> None:
         raise NotImplementedError
 
@@ -108,6 +111,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 +289,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
 
@@ -67,7 +68,9 @@ class TestInit:
         assert start_times_names == {'C1': 0, 'IN1': 0, 'IN2': 0, 'CMUL1': 0, 'CMUL2': 5,
                                      "ADD1": 0, "CMUL3": 7, 'OUT1': 9, 'OUT2': 10}
 
-    def test_slack_normal_latency(self, precedence_sfg_delays):
+
+class TestSlacks:
+    def test_forward_backward_slack_normal_latency(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)
 
@@ -77,3 +80,68 @@ 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
+
+    def test_slacks_normal_latency(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")
+        assert schema.slacks(precedence_sfg_delays.find_by_name("ADD3")[0].graph_id) == (0, 7)
+        assert schema.slacks(precedence_sfg_delays.find_by_name("A2")[0].graph_id) == (16, 0)
+
+
+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")
+        add3_id = precedence_sfg_delays.find_by_name("ADD3")[0].graph_id
+        schema.move_operation(add3_id, 4)
+        assert schema.forward_slack(add3_id) == 3
+        assert schema.backward_slack(add3_id) == 4
+
+        a2_id = precedence_sfg_delays.find_by_name("A2")[0].graph_id
+        assert schema.forward_slack(a2_id) == 4
+        assert schema.backward_slack(a2_id) == 16
+
+        schema.move_operation(a2_id, 2)
+
+        assert schema.forward_slack(add3_id) == 3
+        assert schema.backward_slack(add3_id) == 2
+
+        assert schema.forward_slack(a2_id) == 2
+        assert schema.backward_slack(a2_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)