diff --git a/b_asic/schedule.py b/b_asic/schedule.py
index fa526019979905b2546e291c0653185704aa7fce..b94b9044b67eb645bf89b7af524f1f59055a5f74 100644
--- a/b_asic/schedule.py
+++ b/b_asic/schedule.py
@@ -34,7 +34,7 @@ from b_asic.port import InputPort, OutputPort
from b_asic.process import MemoryVariable, Process
from b_asic.resources import ProcessCollection
from b_asic.signal_flow_graph import SFG
-from b_asic.special_operations import Delay, Output
+from b_asic.special_operations import Delay, Input, Output
# Need RGB from 0 to 1
_EXECUTION_TIME_COLOR = tuple(c / 255 for c in EXECUTION_TIME_COLOR)
@@ -91,9 +91,7 @@ class Schedule:
if schedule_time is None:
self._schedule_time = max_end_time
elif schedule_time < max_end_time:
- raise ValueError(
- f"Too short schedule time. Minimum is {max_end_time}."
- )
+ raise ValueError(f"Too short schedule time. Minimum is {max_end_time}.")
else:
self._schedule_time = schedule_time
@@ -102,9 +100,7 @@ class Schedule:
Return the start time of the operation with the specified by *graph_id*.
"""
if graph_id not in self._start_times:
- raise ValueError(
- f"No operation with graph_id {graph_id} in schedule"
- )
+ raise ValueError(f"No operation with graph_id {graph_id} in schedule")
return self._start_times[graph_id]
def get_max_end_time(self) -> int:
@@ -139,9 +135,7 @@ class Schedule:
slacks
"""
if graph_id not in self._start_times:
- raise ValueError(
- f"No operation with graph_id {graph_id} in schedule"
- )
+ raise ValueError(f"No operation with graph_id {graph_id} in schedule")
slack = sys.maxsize
output_slacks = self._forward_slacks(graph_id)
# Make more pythonic
@@ -194,9 +188,7 @@ class Schedule:
slacks
"""
if graph_id not in self._start_times:
- raise ValueError(
- f"No operation with graph_id {graph_id} in schedule"
- )
+ raise ValueError(f"No operation with graph_id {graph_id} in schedule")
slack = sys.maxsize
input_slacks = self._backward_slacks(graph_id)
# Make more pythonic
@@ -205,9 +197,7 @@ class Schedule:
slack = min(slack, signal_slack)
return slack
- def _backward_slacks(
- self, graph_id: GraphID
- ) -> Dict[InputPort, Dict[Signal, int]]:
+ def _backward_slacks(self, graph_id: GraphID) -> Dict[InputPort, Dict[Signal, int]]:
ret = {}
start_time = self._start_times[graph_id]
op = cast(Operation, self._sfg.find_by_id(graph_id))
@@ -250,9 +240,7 @@ class Schedule:
"""
if graph_id not in self._start_times:
- raise ValueError(
- f"No operation with graph_id {graph_id} in schedule"
- )
+ raise ValueError(f"No operation with graph_id {graph_id} in schedule")
return self.backward_slack(graph_id), self.forward_slack(graph_id)
def print_slacks(self) -> None:
@@ -309,13 +297,11 @@ class Schedule:
factor : int
The time resolution increment.
"""
- self._start_times = {
- k: factor * v for k, v in self._start_times.items()
- }
+ self._start_times = {k: factor * v for k, v in self._start_times.items()}
for graph_id in self._start_times:
- cast(
- Operation, self._sfg.find_by_id(graph_id)
- )._increase_time_resolution(factor)
+ cast(Operation, self._sfg.find_by_id(graph_id))._increase_time_resolution(
+ factor
+ )
self._schedule_time *= factor
return self
@@ -366,13 +352,11 @@ class Schedule:
f"Not possible to decrease resolution with {factor}. Possible"
f" values are {possible_values}"
)
- self._start_times = {
- k: v // factor for k, v in self._start_times.items()
- }
+ self._start_times = {k: v // factor for k, v in self._start_times.items()}
for graph_id in self._start_times:
- cast(
- Operation, self._sfg.find_by_id(graph_id)
- )._decrease_time_resolution(factor)
+ cast(Operation, self._sfg.find_by_id(graph_id))._decrease_time_resolution(
+ factor
+ )
self._schedule_time = self._schedule_time // factor
return self
@@ -388,9 +372,7 @@ class Schedule:
The time to move. If positive move forward, if negative move backward.
"""
if graph_id not in self._start_times:
- raise ValueError(
- f"No operation with graph_id {graph_id} in schedule"
- )
+ raise ValueError(f"No operation with graph_id {graph_id} in schedule")
(backward_slack, forward_slack) = self.slacks(graph_id)
if not -backward_slack <= time <= forward_slack:
@@ -413,15 +395,25 @@ class Schedule:
tmp_prev_available = tmp_usage - new_slack
prev_available = tmp_prev_available % self._schedule_time
laps = new_slack // self._schedule_time
+ source_op = signal.source.operation
if new_usage < prev_available:
print("Incrementing input laps 1")
laps += 1
- if prev_available == 0 and new_usage == 0:
+ if (
+ prev_available == 0
+ and new_usage == 0
+ and (
+ tmp_prev_available > 0
+ or tmp_prev_available == 0
+ and not isinstance(source_op, Input)
+ )
+ ):
print("Incrementing input laps 2")
laps += 1
print(
[
"Input",
+ signal.source.operation,
time,
tmp_start,
signal_slack,
@@ -476,12 +468,8 @@ class Schedule:
while delay_list:
delay_op = cast(Delay, delay_list[0])
delay_input_id = delay_op.input(0).signals[0].graph_id
- delay_output_ids = [
- sig.graph_id for sig in delay_op.output(0).signals
- ]
- self._sfg = cast(
- SFG, self._sfg.remove_operation(delay_op.graph_id)
- )
+ delay_output_ids = [sig.graph_id for sig in delay_op.output(0).signals]
+ self._sfg = cast(SFG, self._sfg.remove_operation(delay_op.graph_id))
for output_id in delay_output_ids:
self._laps[output_id] += 1 + self._laps[delay_input_id]
del self._laps[delay_input_id]
@@ -520,21 +508,16 @@ class Schedule:
for inport in op.inputs:
if len(inport.signals) != 1:
raise ValueError(
- "Error in scheduling, dangling input port"
- " detected."
+ "Error in scheduling, dangling input port detected."
)
if inport.signals[0].source is None:
raise ValueError(
- "Error in scheduling, signal with no source"
- " detected."
+ "Error in scheduling, signal with no source detected."
)
source_port = inport.signals[0].source
source_end_time = None
- if (
- source_port.operation.graph_id
- in non_schedulable_ops
- ):
+ if source_port.operation.graph_id in non_schedulable_ops:
source_end_time = 0
else:
source_op_time = self._start_times[
@@ -559,12 +542,8 @@ class Schedule:
f" {inport.operation.graph_id} has no"
" latency-offset."
)
- op_start_time_from_in = (
- source_end_time - inport.latency_offset
- )
- op_start_time = max(
- op_start_time, op_start_time_from_in
- )
+ op_start_time_from_in = source_end_time - inport.latency_offset
+ op_start_time = max(op_start_time, op_start_time_from_in)
self._start_times[op.graph_id] = op_start_time
for output in self._sfg.find_by_type_name(Output.type_name()):
@@ -625,17 +604,13 @@ class Schedule:
y_location = self._y_locations[graph_id]
if y_location is None:
# Assign the lowest row number not yet in use
- used = set(
- loc for loc in self._y_locations.values() if loc is not None
- )
+ used = set(loc for loc in self._y_locations.values() if loc is not None)
possible = set(range(len(self._start_times))) - used
y_location = min(possible)
self._y_locations[graph_id] = y_location
return operation_gap + y_location * (operation_height + operation_gap)
- def _plot_schedule(
- self, ax: Axes, operation_gap: Optional[float] = None
- ) -> None:
+ def _plot_schedule(self, ax: Axes, operation_gap: Optional[float] = None) -> None:
"""Draw the schedule."""
line_cache = []
@@ -722,9 +697,7 @@ class Schedule:
)
ax.add_patch(pp)
- def _draw_offset_arrow(
- start, end, start_offset, end_offset, name="", laps=0
- ):
+ def _draw_offset_arrow(start, end, start_offset, end_offset, name="", laps=0):
"""Draw an arrow from *start* to *end*, but with an offset."""
_draw_arrow(
[start[0] + start_offset[0], start[1] + start_offset[1]],
@@ -761,24 +734,18 @@ class Schedule:
linewidth=3,
)
ytickpositions.append(y_pos + 0.5)
- yticklabels.append(
- cast(Operation, self._sfg.find_by_id(graph_id)).name
- )
+ yticklabels.append(cast(Operation, self._sfg.find_by_id(graph_id)).name)
for graph_id, op_start_time in self._start_times.items():
op = cast(Operation, self._sfg.find_by_id(graph_id))
out_coordinates = op.get_output_coordinates()
- source_y_pos = self._get_y_position(
- graph_id, operation_gap=operation_gap
- )
+ source_y_pos = self._get_y_position(graph_id, operation_gap=operation_gap)
for output_port in op.outputs:
for output_signal in output_port.signals:
destination = cast(InputPort, output_signal.destination)
destination_op = destination.operation
- destination_start_time = self._start_times[
- destination_op.graph_id
- ]
+ destination_start_time = self._start_times[destination_op.graph_id]
destination_y_pos = self._get_y_position(
destination_op.graph_id, operation_gap=operation_gap
)
@@ -804,9 +771,7 @@ class Schedule:
+ 1
+ (OPERATION_GAP if operation_gap is None else operation_gap)
)
- ax.axis(
- [-1, self._schedule_time + 1, y_position_max, 0]
- ) # Inverted y-axis
+ ax.axis([-1, self._schedule_time + 1, y_position_max, 0]) # Inverted y-axis
ax.xaxis.set_major_locator(MaxNLocator(integer=True))
ax.axvline(
0,
@@ -823,9 +788,7 @@ class Schedule:
"""Reset all the y-locations in the schedule to None"""
self._y_locations = self._y_locations = defaultdict(lambda: None)
- def plot_in_axes(
- self, ax: Axes, operation_gap: Optional[float] = None
- ) -> None:
+ def plot_in_axes(self, ax: Axes, operation_gap: Optional[float] = None) -> None:
"""
Plot the schedule in a :class:`matplotlib.axes.Axes` or subclass.
diff --git a/pyproject.toml b/pyproject.toml
index be37e866e99b80a006e3680bd42109d9ebfaf709..534c0b98f55cd30a19770c496f9187afaa1e4269 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -54,14 +54,14 @@ documentation = "https://da.gitlab-pages.liu.se/B-ASIC/"
[tool.black]
skip-string-normalization = true
preview = true
-line-length = 79
+line-length = 88
exclude = [
"test/test_gui", "b_asic/scheduler_gui/ui_main_window.py"
]
[tool.isort]
profile = "black"
-line_length = 79
+line_length = 88
src_paths = ["b_asic", "test"]
skip = [
"test/test_gui", "b_asic/scheduler_gui/ui_main_window.py"
diff --git a/test/test_schedule.py b/test/test_schedule.py
index 93199f01ebb6585b71bce4abbf2ecb0e3615584a..dcc04c80088ad6a5150840c363e1cd839a2093ec 100644
--- a/test/test_schedule.py
+++ b/test/test_schedule.py
@@ -8,15 +8,13 @@ import pytest
from b_asic.core_operations import Addition, Butterfly, ConstantMultiplication
from b_asic.schedule import Schedule
from b_asic.signal_flow_graph import SFG
-from b_asic.special_operations import Input, Output
+from b_asic.special_operations import Delay, Input, Output
class TestInit:
def test_simple_filter_normal_latency(self, sfg_simple_filter):
sfg_simple_filter.set_latency_of_type(Addition.type_name(), 5)
- sfg_simple_filter.set_latency_of_type(
- ConstantMultiplication.type_name(), 4
- )
+ sfg_simple_filter.set_latency_of_type(ConstantMultiplication.type_name(), 4)
schedule = Schedule(sfg_simple_filter)
@@ -28,13 +26,9 @@ class TestInit:
}
assert schedule.schedule_time == 9
- def test_complicated_single_outputs_normal_latency(
- self, precedence_sfg_delays
- ):
+ def test_complicated_single_outputs_normal_latency(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
- )
+ precedence_sfg_delays.set_latency_of_type(ConstantMultiplication.type_name(), 3)
schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
@@ -88,9 +82,7 @@ class TestInit:
}
assert secondorder_iir_schedule.schedule_time == 21
- def test_complicated_single_outputs_complex_latencies(
- self, precedence_sfg_delays
- ):
+ def test_complicated_single_outputs_complex_latencies(self, precedence_sfg_delays):
precedence_sfg_delays.set_latency_offsets_of_type(
ConstantMultiplication.type_name(), {"in0": 3, "out0": 5}
)
@@ -152,9 +144,7 @@ class TestInit:
assert schedule.schedule_time == 17
- def test_independent_sfg(
- self, sfg_two_inputs_two_outputs_independent_with_cmul
- ):
+ def test_independent_sfg(self, sfg_two_inputs_two_outputs_independent_with_cmul):
schedule = Schedule(
sfg_two_inputs_two_outputs_independent_with_cmul,
scheduling_algorithm="ASAP",
@@ -162,11 +152,9 @@ class TestInit:
start_times_names = {}
for op_id, start_time in schedule._start_times.items():
- op_name = (
- sfg_two_inputs_two_outputs_independent_with_cmul.find_by_id(
- op_id
- ).name
- )
+ op_name = sfg_two_inputs_two_outputs_independent_with_cmul.find_by_id(
+ op_id
+ ).name
start_times_names[op_name] = start_time
assert start_times_names == {
@@ -184,13 +172,9 @@ class TestInit:
class TestSlacks:
- def test_forward_backward_slack_normal_latency(
- self, precedence_sfg_delays
- ):
+ 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
- )
+ precedence_sfg_delays.set_latency_of_type(ConstantMultiplication.type_name(), 3)
schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
assert (
@@ -207,9 +191,7 @@ class TestSlacks:
)
assert (
- schedule.forward_slack(
- precedence_sfg_delays.find_by_name("A2")[0].graph_id
- )
+ schedule.forward_slack(precedence_sfg_delays.find_by_name("A2")[0].graph_id)
== 0
)
assert (
@@ -221,9 +203,7 @@ class TestSlacks:
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
- )
+ precedence_sfg_delays.set_latency_of_type(ConstantMultiplication.type_name(), 3)
schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
assert schedule.slacks(
@@ -237,18 +217,14 @@ class TestSlacks:
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
- )
+ precedence_sfg_delays.set_latency_of_type(ConstantMultiplication.type_name(), 3)
schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
schedule.move_operation(
precedence_sfg_delays.find_by_name("ADD3")[0].graph_id, 4
)
- schedule.move_operation(
- precedence_sfg_delays.find_by_name("A2")[0].graph_id, 2
- )
+ schedule.move_operation(precedence_sfg_delays.find_by_name("A2")[0].graph_id, 2)
start_times_names = {}
for op_id, start_time in schedule._start_times.items():
@@ -271,13 +247,9 @@ class TestRescheduling:
"OUT1": 21,
}
- def test_move_operation_slack_after_rescheduling(
- self, precedence_sfg_delays
- ):
+ 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
- )
+ precedence_sfg_delays.set_latency_of_type(ConstantMultiplication.type_name(), 3)
schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
add3_id = precedence_sfg_delays.find_by_name("ADD3")[0].graph_id
@@ -297,34 +269,97 @@ class TestRescheduling:
assert schedule.forward_slack(a2_id) == 2
assert schedule.backward_slack(a2_id) == 18
- def test_move_operation_incorrect_move_backward(
- self, precedence_sfg_delays
- ):
+ 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
- )
+ precedence_sfg_delays.set_latency_of_type(ConstantMultiplication.type_name(), 3)
schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
- with pytest.raises(ValueError):
+ with pytest.raises(
+ ValueError,
+ match="Operation add4 got incorrect move: -4. Must be between 0 and 7.",
+ ):
schedule.move_operation(
precedence_sfg_delays.find_by_name("ADD3")[0].graph_id, -4
)
- def test_move_operation_incorrect_move_forward(
- self, precedence_sfg_delays
- ):
+ 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
- )
+ precedence_sfg_delays.set_latency_of_type(ConstantMultiplication.type_name(), 3)
schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
- with pytest.raises(ValueError):
+ with pytest.raises(
+ ValueError,
+ match="Operation add4 got incorrect move: 10. Must be between 0 and 7.",
+ ):
schedule.move_operation(
precedence_sfg_delays.find_by_name("ADD3")[0].graph_id, 10
)
+ def test_move_operation_acc(self):
+ in0 = Input()
+ d = Delay()
+ a = d + in0
+ out0 = Output(a)
+ d << a
+ sfg = SFG([in0], [out0])
+ sfg.set_latency_of_type(Addition.type_name(), 1)
+ schedule = Schedule(sfg, cyclic=True)
+
+ # Check initial conditions
+ assert schedule.laps[sfg.find_by_id("add1").input(0).signals[0].graph_id] == 1
+ assert schedule.laps[sfg.find_by_id("add1").input(1).signals[0].graph_id] == 0
+ assert schedule._start_times["add1"] == 0
+ assert schedule.laps[sfg.find_by_id("out1").input(0).signals[0].graph_id] == 0
+ assert schedule._start_times["out1"] == 1
+
+ # Move and scheduling algorithm behaves differently
+ schedule.move_operation("out1", 0)
+ assert schedule.laps[sfg.find_by_id("out1").input(0).signals[0].graph_id] == 1
+ assert schedule.laps[sfg.find_by_id("add1").input(0).signals[0].graph_id] == 1
+ assert schedule.laps[sfg.find_by_id("add1").input(1).signals[0].graph_id] == 0
+ assert schedule._start_times["out1"] == 0
+ assert schedule._start_times["add1"] == 0
+
+ # Increase schedule time
+ schedule.set_schedule_time(2)
+ assert schedule.laps[sfg.find_by_id("out1").input(0).signals[0].graph_id] == 1
+ assert schedule.laps[sfg.find_by_id("add1").input(0).signals[0].graph_id] == 1
+ assert schedule.laps[sfg.find_by_id("add1").input(1).signals[0].graph_id] == 0
+ assert schedule._start_times["out1"] == 0
+ assert schedule._start_times["add1"] == 0
+
+ # Move out one time unit
+ schedule.move_operation("out1", 1)
+ assert schedule.laps[sfg.find_by_id("out1").input(0).signals[0].graph_id] == 1
+ assert schedule.laps[sfg.find_by_id("add1").input(0).signals[0].graph_id] == 1
+ assert schedule.laps[sfg.find_by_id("add1").input(1).signals[0].graph_id] == 0
+ assert schedule._start_times["out1"] == 1
+ assert schedule._start_times["add1"] == 0
+
+ # Move add one time unit
+ schedule.move_operation("add1", 1)
+ assert schedule.laps[sfg.find_by_id("add1").input(0).signals[0].graph_id] == 1
+ assert schedule.laps[sfg.find_by_id("add1").input(1).signals[0].graph_id] == 0
+ assert schedule.laps[sfg.find_by_id("out1").input(0).signals[0].graph_id] == 1
+ assert schedule._start_times["add1"] == 1
+ assert schedule._start_times["out1"] == 1
+
+ # Move out back one time unit
+ schedule.move_operation("out1", -1)
+ assert schedule.laps[sfg.find_by_id("out1").input(0).signals[0].graph_id] == 1
+ assert schedule._start_times["out1"] == 0
+ assert schedule.laps[sfg.find_by_id("add1").input(0).signals[0].graph_id] == 1
+ assert schedule.laps[sfg.find_by_id("add1").input(1).signals[0].graph_id] == 0
+ assert schedule._start_times["add1"] == 1
+
+ # Move add back one time unit
+ schedule.move_operation("add1", -1)
+ assert schedule.laps[sfg.find_by_id("add1").input(0).signals[0].graph_id] == 1
+ assert schedule.laps[sfg.find_by_id("add1").input(1).signals[0].graph_id] == 0
+ assert schedule.laps[sfg.find_by_id("out1").input(0).signals[0].graph_id] == 1
+ assert schedule._start_times["add1"] == 0
+ assert schedule._start_times["out1"] == 0
+
class TestTimeResolution:
def test_increase_time_resolution(
@@ -341,11 +376,9 @@ class TestTimeResolution:
start_times_names = {}
for op_id, start_time in schedule._start_times.items():
- op_name = (
- sfg_two_inputs_two_outputs_independent_with_cmul.find_by_id(
- op_id
- ).name
- )
+ op_name = sfg_two_inputs_two_outputs_independent_with_cmul.find_by_id(
+ op_id
+ ).name
start_times_names[op_name] = start_time
assert start_times_names == {
@@ -377,11 +410,9 @@ class TestTimeResolution:
start_times_names = {}
for op_id, start_time in schedule._start_times.items():
- op_name = (
- sfg_two_inputs_two_outputs_independent_with_cmul.find_by_id(
- op_id
- ).name
- )
+ op_name = sfg_two_inputs_two_outputs_independent_with_cmul.find_by_id(
+ op_id
+ ).name
start_times_names[op_name] = start_time
assert start_times_names == {
@@ -418,11 +449,9 @@ class TestTimeResolution:
start_times_names = {}
for op_id, start_time in schedule._start_times.items():
- op_name = (
- sfg_two_inputs_two_outputs_independent_with_cmul.find_by_id(
- op_id
- ).name
- )
+ op_name = sfg_two_inputs_two_outputs_independent_with_cmul.find_by_id(
+ op_id
+ ).name
start_times_names[op_name] = start_time
assert start_times_names == {
@@ -437,19 +466,15 @@ class TestTimeResolution:
"OUT2": 60,
}
- with pytest.raises(
- ValueError, match="Not possible to decrease resolution"
- ):
+ with pytest.raises(ValueError, match="Not possible to decrease resolution"):
schedule.decrease_time_resolution(4)
schedule.decrease_time_resolution(3)
start_times_names = {}
for op_id, start_time in schedule._start_times.items():
- op_name = (
- sfg_two_inputs_two_outputs_independent_with_cmul.find_by_id(
- op_id
- ).name
- )
+ op_name = sfg_two_inputs_two_outputs_independent_with_cmul.find_by_id(
+ op_id
+ ).name
start_times_names[op_name] = start_time
assert start_times_names == {
@@ -491,9 +516,7 @@ class TestErrors:
def test_no_output_latency(self):
in1 = Input()
in2 = Input()
- bfly = Butterfly(
- in1, in2, latency_offsets={"in0": 4, "in1": 2, "out0": 10}
- )
+ bfly = Butterfly(in1, in2, latency_offsets={"in0": 4, "in1": 2, "out0": 10})
out1 = Output(bfly.output(0))
out2 = Output(bfly.output(1))
sfg = SFG([in1, in2], [out1, out2])
@@ -504,9 +527,7 @@ class TestErrors:
Schedule(sfg)
in1 = Input()
in2 = Input()
- bfly1 = Butterfly(
- in1, in2, latency_offsets={"in0": 4, "in1": 2, "out1": 10}
- )
+ bfly1 = Butterfly(in1, in2, latency_offsets={"in0": 4, "in1": 2, "out1": 10})
bfly2 = Butterfly(
bfly1.output(0),
bfly1.output(1),
@@ -523,12 +544,8 @@ class TestErrors:
def test_too_short_schedule_time(self, sfg_simple_filter):
sfg_simple_filter.set_latency_of_type(Addition.type_name(), 5)
- sfg_simple_filter.set_latency_of_type(
- ConstantMultiplication.type_name(), 4
- )
- with pytest.raises(
- ValueError, match="Too short schedule time. Minimum is 9."
- ):
+ sfg_simple_filter.set_latency_of_type(ConstantMultiplication.type_name(), 4)
+ with pytest.raises(ValueError, match="Too short schedule time. Minimum is 9."):
Schedule(sfg_simple_filter, schedule_time=3)
schedule = Schedule(sfg_simple_filter)
@@ -540,9 +557,7 @@ class TestErrors:
def test_incorrect_scheduling_algorithm(self, sfg_simple_filter):
sfg_simple_filter.set_latency_of_type(Addition.type_name(), 1)
- sfg_simple_filter.set_latency_of_type(
- ConstantMultiplication.type_name(), 2
- )
+ sfg_simple_filter.set_latency_of_type(ConstantMultiplication.type_name(), 2)
with pytest.raises(
NotImplementedError, match="No algorithm with name: foo defined."
):