diff --git a/README.md b/README.md
index b2972f30828f19038e5efe612831f989b8f5ffd5..d28399ce800d0240881c14405c892eb65d23b072 100644
--- a/README.md
+++ b/README.md
@@ -29,6 +29,15 @@ To run the test suite, the following additional packages are required:
* pytest
* pytest-cov (for testing with coverage)
+To build a binary distribution, the following additional packages are required:
+* Python:
+ * wheel
+
+To run the test suite, the following additional packages are required:
+* Python:
+ * pytest
+ * pytest-cov (for testing with coverage)
+
### Using CMake directly
How to build using CMake.
diff --git a/b_asic/GUI/drag_button.py b/b_asic/GUI/drag_button.py
index 80bfcf335ca10c54a820749bdc2225d18cbdd060..fe78f46323400555fce5687cc2459009e61c0a4a 100644
--- a/b_asic/GUI/drag_button.py
+++ b/b_asic/GUI/drag_button.py
@@ -121,11 +121,16 @@ class DragButton(QPushButton):
_signals = []
for signal, ports in self._window.signalPortDict.items():
- if any([port in self._window.portDict[self] for port in ports]):
+ if any(map(lambda port: set(port).intersection(set(self._window.portDict[self])), ports)):
self._window.logger.info(f"Removed signal with name: {signal.signal.name} to/from operation: {self.operation.name}.")
signal.remove()
_signals.append(signal)
+ if self in self._window.opToSFG:
+ self._window.logger.info(f"Operation detected in existing sfg, removing sfg with name: {self._window.opToSFG[self].name}.")
+ self._window.opToSFG = {op: self._window.opToSFG[op] for op in self._window.opToSFG if self._window.opToSFG[op] is self._window.opToSFG[self]}
+ del self._window.sfg_dict[self._window.opToSFG[self].name]
+
for signal in _signals:
del self._window.signalPortDict[signal]
diff --git a/b_asic/GUI/main_window.py b/b_asic/GUI/main_window.py
index a39e679cf30c23697685fc46a117de248ba7612a..b9841b319a2d688dfa2b58b13b6c587bcf8443e1 100644
--- a/b_asic/GUI/main_window.py
+++ b/b_asic/GUI/main_window.py
@@ -61,6 +61,7 @@ class MainWindow(QMainWindow):
self.pressed_operations = []
self.portDict = dict()
self.signalPortDict = dict()
+ self.opToSFG = dict()
self.pressed_ports = []
self.sfg_dict = dict()
self._window = self
@@ -266,6 +267,7 @@ class MainWindow(QMainWindow):
for op in self.pressed_operations:
op.setToolTip(sfg.name)
+ self.opToSFG[op] = sfg
self.sfg_dict[sfg.name] = sfg
@@ -396,7 +398,7 @@ class MainWindow(QMainWindow):
def _connect_button(self, event):
if len(self.pressed_ports) < 2:
- self.logger.warn("Can't connect less than two ports. Please select more.")
+ self.logger.warning("Can't connect less than two ports. Please select more.")
return
for i in range(len(self.pressed_ports) - 1):
diff --git a/b_asic/GUI/properties_window.py b/b_asic/GUI/properties_window.py
index 22fdf7d2e9329571e98e38e4aad242b6fd50733e..124364a14de3184261c4e7d96de3ea8c5706b79a 100644
--- a/b_asic/GUI/properties_window.py
+++ b/b_asic/GUI/properties_window.py
@@ -1,5 +1,5 @@
from PySide2.QtWidgets import QDialog, QLineEdit, QPushButton, QVBoxLayout, QHBoxLayout,\
-QLabel, QCheckBox
+QLabel, QCheckBox, QGridLayout
from PySide2.QtCore import Qt
from PySide2.QtGui import QIntValidator
@@ -17,6 +17,7 @@ class PropertiesWindow(QDialog):
self.edit_name = QLineEdit(self.operation.operation_path_name)
self.name_layout.addWidget(self.name_label)
self.name_layout.addWidget(self.edit_name)
+ self.latency_fields = dict()
self.vertical_layout = QVBoxLayout()
self.vertical_layout.addLayout(self.name_layout)
@@ -43,6 +44,76 @@ class PropertiesWindow(QDialog):
self.show_name_layout.addWidget(self.check_show_name)
self.vertical_layout.addLayout(self.show_name_layout)
+ if self.operation.operation.input_count > 0:
+ self.latency_layout = QHBoxLayout()
+ self.latency_label = QLabel("Set Latency For Input Ports (-1 for None):")
+ self.latency_layout.addWidget(self.latency_label)
+ self.vertical_layout.addLayout(self.latency_layout)
+
+ input_grid = QGridLayout()
+ x, y = 0, 0
+ for i in range(self.operation.operation.input_count):
+ input_layout = QHBoxLayout()
+ input_layout.addStretch()
+ if i % 2 == 0 and i > 0:
+ x += 1
+ y = 0
+
+ input_label = QLabel("in" + str(i))
+ input_layout.addWidget(input_label)
+ input_value = QLineEdit()
+ try:
+ input_value.setPlaceholderText(str(self.operation.operation.latency))
+ except ValueError:
+ input_value.setPlaceholderText("-1")
+ int_valid = QIntValidator()
+ int_valid.setBottom(-1)
+ input_value.setValidator(int_valid)
+ input_value.setFixedWidth(50)
+ self.latency_fields["in" + str(i)] = input_value
+ input_layout.addWidget(input_value)
+ input_layout.addStretch()
+ input_layout.setSpacing(10)
+ input_grid.addLayout(input_layout, x, y)
+ y += 1
+
+ self.vertical_layout.addLayout(input_grid)
+
+ if self.operation.operation.output_count > 0:
+ self.latency_layout = QHBoxLayout()
+ self.latency_label = QLabel("Set Latency For Output Ports (-1 for None):")
+ self.latency_layout.addWidget(self.latency_label)
+ self.vertical_layout.addLayout(self.latency_layout)
+
+ input_grid = QGridLayout()
+ x, y = 0, 0
+ for i in range(self.operation.operation.output_count):
+ input_layout = QHBoxLayout()
+ input_layout.addStretch()
+ if i % 2 == 0 and i > 0:
+ x += 1
+ y = 0
+
+ input_label = QLabel("out" + str(i))
+ input_layout.addWidget(input_label)
+ input_value = QLineEdit()
+ try:
+ input_value.setPlaceholderText(str(self.operation.operation.latency))
+ except ValueError:
+ input_value.setPlaceholderText("-1")
+ int_valid = QIntValidator()
+ int_valid.setBottom(-1)
+ input_value.setValidator(int_valid)
+ input_value.setFixedWidth(50)
+ self.latency_fields["out" + str(i)] = input_value
+ input_layout.addWidget(input_value)
+ input_layout.addStretch()
+ input_layout.setSpacing(10)
+ input_grid.addLayout(input_layout, x, y)
+ y += 1
+
+ self.vertical_layout.addLayout(input_grid)
+
self.ok = QPushButton("OK")
self.ok.clicked.connect(self.save_properties)
self.vertical_layout.addWidget(self.ok)
@@ -60,4 +131,7 @@ class PropertiesWindow(QDialog):
else:
self.operation.label.setOpacity(0)
self.operation.is_show_name = False
+
+ self.operation.operation.set_latency_offsets({port: int(self.latency_fields[port].text()) if self.latency_fields[port].text() and int(self.latency_fields[port].text()) > 0 else None for port in self.latency_fields})
+
self.reject()
\ No newline at end of file
diff --git a/b_asic/GUI/simulate_sfg_window.py b/b_asic/GUI/simulate_sfg_window.py
index 48e584dc87e045322f5d92946b431d32d7e180b5..161e569aed7a9e7571e8dfacc451d18eb5f6224c 100644
--- a/b_asic/GUI/simulate_sfg_window.py
+++ b/b_asic/GUI/simulate_sfg_window.py
@@ -16,6 +16,7 @@ class SimulateSFGWindow(QDialog):
self._window = window
self.properties = dict()
self.sfg_to_layout = dict()
+ self.input_fields = dict()
self.setWindowFlags(Qt.WindowTitleHint | Qt.WindowCloseButtonHint)
self.setWindowTitle("Simulate SFG")
@@ -36,32 +37,50 @@ class SimulateSFGWindow(QDialog):
spin_box.setRange(0, 2147483647)
options_layout.addRow("Iteration Count: ", spin_box)
- check_box = QCheckBox()
- options_layout.addRow("Plot Results: ", check_box)
-
- check_box = QCheckBox()
- options_layout.addRow("Get All Results: ", check_box)
-
- input_layout = QVBoxLayout()
- input_label = QLabel("Input Values: ")
- input_layout.addWidget(input_label)
- input_grid = QGridLayout()
- x, y = 0, 0
- for i in range(sfg.input_count):
- if i % 2 == 0 and i > 0:
- x += 1
- y = 0
-
- input_value = QLineEdit()
- input_value.setPlaceholderText(str(i))
- input_value.setValidator(QIntValidator())
- input_value.setFixedWidth(50)
- input_grid.addWidget(input_value, x, y)
- y += 1
-
- input_layout.addLayout(input_grid)
+ check_box_plot = QCheckBox()
+ options_layout.addRow("Plot Results: ", check_box_plot)
+
+ check_box_all = QCheckBox()
+ options_layout.addRow("Get All Results: ", check_box_all)
+
sfg_layout.addLayout(options_layout)
- sfg_layout.addLayout(input_layout)
+
+ self.input_fields[sfg] = {
+ "iteration_count": spin_box,
+ "show_plot": check_box_plot,
+ "all_results": check_box_all,
+ "input_values": []
+ }
+
+ if sfg.input_count > 0:
+ input_label = QHBoxLayout()
+ input_label = QLabel("Input Values:")
+ options_layout.addRow(input_label)
+
+ input_grid = QGridLayout()
+ x, y = 0, 0
+ for i in range(sfg.input_count):
+ input_layout = QHBoxLayout()
+ input_layout.addStretch()
+ if i % 2 == 0 and i > 0:
+ x += 1
+ y = 0
+
+ input_label = QLabel("in" + str(i))
+ input_layout.addWidget(input_label)
+ input_value = QLineEdit()
+ input_value.setPlaceholderText("0")
+ input_value.setValidator(QIntValidator())
+ input_value.setFixedWidth(50)
+ input_layout.addWidget(input_value)
+ input_layout.addStretch()
+ input_layout.setSpacing(10)
+ input_grid.addLayout(input_layout, x, y)
+
+ self.input_fields[sfg]["input_values"].append(input_value)
+ y += 1
+
+ sfg_layout.addLayout(input_grid)
frame = QFrame()
frame.setFrameShape(QFrame.HLine)
@@ -72,21 +91,17 @@ class SimulateSFGWindow(QDialog):
self.dialog_layout.addLayout(sfg_layout)
def save_properties(self):
- for sfg, sfg_row in self.sfg_to_layout.items():
- _option_values = sfg_row.children()[0]
- _input_values = sfg_row.children()[1].children()[0]
- _, spin_box, _, check_box_plot, _, check_box_all = map(lambda x: _option_values.itemAt(x).widget(), range(_option_values.count()))
- input_values = map(lambda x: _input_values.itemAt(x).widget(), range(_input_values.count()))
- input_values = [int(widget.text()) if widget.text() else 0 for widget in input_values]
+ for sfg, properties in self.input_fields.items():
+
self.properties[sfg] = {
- "iteration_count": spin_box.value(),
- "show_plot": check_box_plot.isChecked(),
- "all_results": check_box_all.isChecked(),
- "input_values": input_values
+ "iteration_count": self.input_fields[sfg]["iteration_count"].value(),
+ "show_plot": self.input_fields[sfg]["show_plot"].isChecked(),
+ "all_results": self.input_fields[sfg]["all_results"].isChecked(),
+ "input_values": [int(widget.text()) if widget.text() else 0 for widget in self.input_fields[sfg]["input_values"]]
}
# If we plot we should also print the entire data, since you can't really interact with the graph.
- if check_box_plot.isChecked():
+ if self.properties[sfg]["show_plot"]:
self.properties[sfg]["all_results"] = True
self.accept()
diff --git a/b_asic/__init__.py b/b_asic/__init__.py
index 2a184bcae98fc035d609efebc13f9bebe2368b40..55a7250c6c1f4f53191a3bf834de9c0bf2e61fd2 100644
--- a/b_asic/__init__.py
+++ b/b_asic/__init__.py
@@ -11,3 +11,4 @@ from b_asic.signal import *
from b_asic.simulation import *
from b_asic.special_operations import *
from b_asic.save_load_structure import *
+from b_asic.schema import *
diff --git a/b_asic/core_operations.py b/b_asic/core_operations.py
index a70c86b7d966197850604ad7a269242d364eca36..9a11d91943e9621e43daa3c88ddfa67d3441060c 100644
--- a/b_asic/core_operations.py
+++ b/b_asic/core_operations.py
@@ -4,7 +4,7 @@ TODO: More info.
"""
from numbers import Number
-from typing import Optional
+from typing import Optional, Dict
from numpy import conjugate, sqrt, abs as np_abs
from b_asic.port import SignalSourceProvider, InputPort, OutputPort
@@ -44,9 +44,9 @@ class Addition(AbstractOperation):
TODO: More info.
"""
- def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
- super().__init__(input_count=2, output_count=1,
- name=name, input_sources=[src0, src1])
+ def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
+ super().__init__(input_count=2, output_count=1, name=name, input_sources=[src0, src1],
+ latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
@@ -61,9 +61,9 @@ class Subtraction(AbstractOperation):
TODO: More info.
"""
- def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
- super().__init__(input_count=2, output_count=1,
- name=name, input_sources=[src0, src1])
+ def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
+ super().__init__(input_count=2, output_count=1, name=name, input_sources=[src0, src1],
+ latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
@@ -78,9 +78,9 @@ class Multiplication(AbstractOperation):
TODO: More info.
"""
- def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
- super().__init__(input_count=2, output_count=1,
- name=name, input_sources=[src0, src1])
+ def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
+ super().__init__(input_count=2, output_count=1, name=name, input_sources=[src0, src1],
+ latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
@@ -95,9 +95,9 @@ class Division(AbstractOperation):
TODO: More info.
"""
- def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
- super().__init__(input_count=2, output_count=1,
- name=name, input_sources=[src0, src1])
+ def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
+ super().__init__(input_count=2, output_count=1, name=name, input_sources=[src0, src1],
+ latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
@@ -112,9 +112,9 @@ class Min(AbstractOperation):
TODO: More info.
"""
- def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
- super().__init__(input_count=2, output_count=1,
- name=name, input_sources=[src0, src1])
+ def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
+ super().__init__(input_count=2, output_count=1, name=name, input_sources=[src0, src1],
+ latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
@@ -131,9 +131,9 @@ class Max(AbstractOperation):
TODO: More info.
"""
- def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
- super().__init__(input_count=2, output_count=1,
- name=name, input_sources=[src0, src1])
+ def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
+ super().__init__(input_count=2, output_count=1, name=name, input_sources=[src0, src1],
+ latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
@@ -150,9 +150,9 @@ class SquareRoot(AbstractOperation):
TODO: More info.
"""
- def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
- super().__init__(input_count=1, output_count=1,
- name=name, input_sources=[src0])
+ def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
+ super().__init__(input_count=1, output_count=1, name=name, input_sources=[src0],
+ latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
@@ -167,9 +167,9 @@ class ComplexConjugate(AbstractOperation):
TODO: More info.
"""
- def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
- super().__init__(input_count=1, output_count=1,
- name=name, input_sources=[src0])
+ def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
+ super().__init__(input_count=1, output_count=1, name=name, input_sources=[src0],
+ latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
@@ -184,9 +184,9 @@ class Absolute(AbstractOperation):
TODO: More info.
"""
- def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
- super().__init__(input_count=1, output_count=1,
- name=name, input_sources=[src0])
+ def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
+ super().__init__(input_count=1, output_count=1, name=name, input_sources=[src0],
+ latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
@@ -201,9 +201,9 @@ class ConstantMultiplication(AbstractOperation):
TODO: More info.
"""
- def __init__(self, value: Number = 0, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
- super().__init__(input_count=1, output_count=1,
- name=name, input_sources=[src0])
+ def __init__(self, value: Number = 0, src0: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
+ super().__init__(input_count=1, output_count=1, name=name, input_sources=[src0],
+ latency=latency, latency_offsets=latency_offsets)
self.set_param("value", value)
@classmethod
@@ -230,9 +230,9 @@ class Butterfly(AbstractOperation):
TODO: More info.
"""
- def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
- super().__init__(input_count=2, output_count=2,
- name=name, input_sources=[src0, src1])
+ def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
+ super().__init__(input_count=2, output_count=2, name=name, input_sources=[src0, src1],
+ latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
@@ -241,13 +241,15 @@ class Butterfly(AbstractOperation):
def evaluate(self, a, b):
return a + b, a - b
+
class MAD(AbstractOperation):
"""Multiply-and-add operation.
TODO: More info.
"""
- def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, src2: Optional[SignalSourceProvider] = None, name: Name = ""):
- super().__init__(input_count = 3, output_count = 1, name = name, input_sources = [src0, src1, src2])
+ def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, src2: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
+ super().__init__(input_count=3, output_count=1, name=name, input_sources=[src0, src1, src2],
+ latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
diff --git a/b_asic/operation.py b/b_asic/operation.py
index 02ba1aa50682448e931a0694d24e03e20eadd399..91ef5b89a59054527aeba0939e27b78502bd4b57 100644
--- a/b_asic/operation.py
+++ b/b_asic/operation.py
@@ -3,16 +3,17 @@ B-ASIC Operation Module.
TODO: More info.
"""
+from b_asic.signal import Signal
+from b_asic.port import SignalSourceProvider, InputPort, OutputPort
+from b_asic.graph_component import GraphComponent, AbstractGraphComponent, Name
+import itertools as it
+from math import trunc
import collections
from abc import abstractmethod
from numbers import Number
-from typing import NewType, List, Sequence, Iterable, Mapping, MutableMapping, Optional, Any, Set, Union
-from math import trunc
+from typing import NewType, List, Dict, Sequence, Iterable, Mapping, MutableMapping, Optional, Any, Set, Union
-from b_asic.graph_component import GraphComponent, AbstractGraphComponent, Name
-from b_asic.port import SignalSourceProvider, InputPort, OutputPort
-from b_asic.signal import Signal
OutputKey = NewType("OutputKey", str)
OutputMap = Mapping[OutputKey, Optional[Number]]
@@ -193,6 +194,39 @@ class Operation(GraphComponent, SignalSourceProvider):
"""
raise NotImplementedError
+ @property
+ @abstractmethod
+ def latency(self) -> int:
+ """Get the latency of the operation, which is the longest time it takes from one of
+ the operations inputport to one of the operations outputport.
+ """
+ raise NotImplementedError
+
+ @property
+ @abstractmethod
+ def latency_offsets(self) -> Sequence[Sequence[int]]:
+ """Get a nested list with all the operations ports latency-offsets, the first list contains the
+ latency-offsets of the operations input ports, the second list contains the latency-offsets of
+ the operations output ports.
+ """
+ raise NotImplementedError
+
+ @abstractmethod
+ def set_latency(self, latency: int) -> None:
+ """Sets the latency of the operation to the specified integer value by setting the
+ latency-offsets of operations input ports to 0 and the latency-offsets of the operations
+ output ports to the specified value. The latency cannot be a negative integers.
+ """
+ raise NotImplementedError
+
+ @abstractmethod
+ def set_latency_offsets(self, latency_offsets: Dict[str, int]) -> None:
+ """Sets the latency-offsets for the operations ports specified in the latency_offsets dictionary.
+ The latency offsets dictionary should be {'in0': 2, 'out1': 4} if you want to set the latency offset
+ for the inport port with index 0 to 2, and the latency offset of the output port with index 1 to 4.
+ """
+ raise NotImplementedError
+
class AbstractOperation(Operation, AbstractGraphComponent):
"""Generic abstract operation class which most implementations will derive from.
@@ -202,7 +236,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
_input_ports: List[InputPort]
_output_ports: List[OutputPort]
- def __init__(self, input_count: int, output_count: int, name: Name = "", input_sources: Optional[Sequence[Optional[SignalSourceProvider]]] = None):
+ def __init__(self, input_count: int, output_count: int, name: Name = "", input_sources: Optional[Sequence[Optional[SignalSourceProvider]]] = None, latency: int = None, latency_offsets: Dict[str, int] = None):
super().__init__(name)
self._input_ports = [InputPort(self, i) for i in range(input_count)]
@@ -218,6 +252,22 @@ class AbstractOperation(Operation, AbstractGraphComponent):
if src is not None:
self._input_ports[i].connect(src.source)
+ ports_without_latency_offset = set(([f"in{i}" for i in range(self.input_count)] +
+ [f"out{i}" for i in range(self.output_count)]))
+
+ if latency_offsets is not None:
+ self.set_latency_offsets(latency_offsets)
+
+ if latency is not None:
+ # Set the latency of the rest of ports with no latency_offset.
+ assert latency >= 0, "Negative latency entered"
+ for inp in self.inputs:
+ if inp.latency_offset is None:
+ inp.latency_offset = 0
+ for outp in self.outputs:
+ if outp.latency_offset is None:
+ outp.latency_offset = latency
+
@abstractmethod
def evaluate(self, *inputs) -> Any: # pylint: disable=arguments-differ
"""Evaluate the operation and generate a list of output values given a list of input values."""
@@ -433,6 +483,14 @@ class AbstractOperation(Operation, AbstractGraphComponent):
return SFG(inputs=inputs, outputs=outputs)
+ def copy_component(self, *args, **kwargs) -> Operation:
+ new_component = super().copy_component(*args, **kwargs)
+ for i, inp in enumerate(self.inputs):
+ new_component.input(i).latency_offset = inp.latency_offset
+ for i, outp in enumerate(self.outputs):
+ new_component.output(i).latency_offset = outp.latency_offset
+ return new_component
+
def inputs_required_for_output(self, output_index: int) -> Iterable[int]:
if output_index < 0 or output_index >= self.output_count:
raise IndexError(f"Output index out of range (expected 0-{self.output_count - 1}, got {output_index})")
@@ -483,3 +541,44 @@ class AbstractOperation(Operation, AbstractGraphComponent):
else:
args.append(input_values[i])
return args
+
+ @property
+ def latency(self) -> int:
+ if None in [inp.latency_offset for inp in self.inputs] or None in [outp.latency_offset for outp in self.outputs]:
+ raise ValueError("All native offsets have to set to a non-negative value to calculate the latency.")
+
+ return max(((outp.latency_offset - inp.latency_offset) for outp, inp in it.product(self.outputs, self.inputs)))
+
+ def set_latency(self, latency: int) -> None:
+ assert latency >= 0, "Negative latency entered."
+ for inport in self.inputs:
+ inport.latency_offset = 0
+ for outport in self.outputs:
+ outport.latency_offset = latency
+
+ @property
+ def latency_offsets(self) -> Sequence[Sequence[int]]:
+ latency_offsets = dict()
+
+ for i, inp in enumerate(self.inputs):
+ latency_offsets["in" + str(i)] = inp.latency_offset
+
+ for i, outp in enumerate(self.outputs):
+ latency_offsets["out" + str(i)] = outp.latency_offset
+
+ return latency_offsets
+ #return ([inp.latency_offset for inp in self.inputs], [outp.latency_offset for outp in self.outputs])
+
+ def set_latency_offsets(self, latency_offsets: Dict[str, int]) -> None:
+ for port_str, latency_offset in latency_offsets.items():
+ port_str = port_str.lower()
+ if port_str.startswith("in"):
+ index_str = port_str[2:]
+ assert index_str.isdigit(), "Incorrectly formatted index in string, expected 'in' + index"
+ self.input(int(index_str)).latency_offset = latency_offset
+ elif port_str.startswith("out"):
+ index_str = port_str[3:]
+ assert index_str.isdigit(), "Incorrectly formatted index in string, expected 'out' + index"
+ self.output(int(index_str)).latency_offset = latency_offset
+ else:
+ raise ValueError("Incorrectly formatted string, expected 'in' + index or 'out' + index")
diff --git a/b_asic/port.py b/b_asic/port.py
index 20783d5df0962b034aee2b6e934255a9fc9cd6e6..fb3f64177e74f2623d02284243a03529cf05b6e4 100644
--- a/b_asic/port.py
+++ b/b_asic/port.py
@@ -32,6 +32,18 @@ class Port(ABC):
"""Return the index of the port."""
raise NotImplementedError
+ @property
+ @abstractmethod
+ def latency_offset(self) -> int:
+ """Get the latency_offset of the port."""
+ raise NotImplementedError
+
+ @latency_offset.setter
+ @abstractmethod
+ def latency_offset(self, latency_offset: int) -> None:
+ """Set the latency_offset of the port to the integer specified value."""
+ raise NotImplementedError
+
@property
@abstractmethod
def signal_count(self) -> int:
@@ -76,10 +88,12 @@ class AbstractPort(Port):
_operation: "Operation"
_index: int
+ _latency_offset: Optional[int]
- def __init__(self, operation: "Operation", index: int):
+ def __init__(self, operation: "Operation", index: int, latency_offset: int = None):
self._operation = operation
self._index = index
+ self._latency_offset = latency_offset
@property
def operation(self) -> "Operation":
@@ -89,6 +103,14 @@ class AbstractPort(Port):
def index(self) -> int:
return self._index
+ @property
+ def latency_offset(self) -> int:
+ return self._latency_offset
+
+ @latency_offset.setter
+ def latency_offset(self, latency_offset: int):
+ self._latency_offset = latency_offset
+
class SignalSourceProvider(ABC):
"""Signal source provider interface.
diff --git a/b_asic/save_load_structure.py b/b_asic/save_load_structure.py
index e0f6bec387b4b95c383e328e5d2eac3f0673e285..5311d3fe72b1bb711b21c0aa3070b538db81d2d6 100644
--- a/b_asic/save_load_structure.py
+++ b/b_asic/save_load_structure.py
@@ -24,7 +24,7 @@ def sfg_to_python(sfg: SFG, counter: int = 0, suffix: str = None) -> str:
def kwarg_unpacker(comp: GraphComponent, params=None) -> str:
if params is None:
- params_filtered = {attr: getattr(op, attr) for attr in signature(op.__init__).parameters if hasattr(op, attr)}
+ params_filtered = {attr: getattr(op, attr) for attr in signature(op.__init__).parameters if attr != "latency" and hasattr(op, attr)}
params = {attr: getattr(op, attr) if not isinstance(getattr(op, attr), str) else f'"{getattr(op, attr)}"' for attr in params_filtered}
return ", ".join([f"{param[0]}={param[1]}" for param in params.items()])
diff --git a/b_asic/schema.py b/b_asic/schema.py
new file mode 100644
index 0000000000000000000000000000000000000000..157ed3b8ec57a35ffe66f1de476e2d77c3e28abd
--- /dev/null
+++ b/b_asic/schema.py
@@ -0,0 +1,107 @@
+"""@package docstring
+This module contains the Schema class.
+TODO: More info
+"""
+
+from typing import Dict, List
+
+from b_asic.signal_flow_graph import SFG
+from b_asic.graph_component import GraphID
+from b_asic.operation import Operation
+
+
+class Schema:
+ """A class that represents an SFG with scheduled Operations."""
+
+ _sfg: SFG
+ _start_times: Dict[GraphID, int]
+ _laps: Dict[GraphID, List[int]]
+ _schedule_time: int
+ _cyclic: bool
+ _resolution: int
+
+ def __init__(self, sfg: SFG, schedule_time: int = None, cyclic: bool = False, resolution: int = 1, scheduling_alg: str = "ASAP"):
+
+ self._sfg = sfg
+ self._start_times = dict()
+ self._laps = dict()
+ self._cyclic = cyclic
+ self._resolution = resolution
+
+ if scheduling_alg == "ASAP":
+ self._schedule_asap()
+ else:
+ raise NotImplementedError(f"No algorithm with name: {scheduling_alg} defined.")
+
+ max_end_time = 0
+ for op_id, op_start_time in self._start_times.items():
+ op = self._sfg.find_by_id(op_id)
+ for outport in op.outputs:
+ max_end_time = max(max_end_time, op_start_time + outport.latency_offset)
+
+ if not self._cyclic:
+ if schedule_time is None:
+ self._schedule_time = max_end_time
+ elif schedule_time < max_end_time:
+ raise ValueError("Too short schedule time for non-cyclic Schedule entered.")
+ else:
+ self._schedule_time = schedule_time
+
+ def start_time_of_operation(self, op_id: GraphID):
+ """Get the start time of the operation with the specified by the op_id."""
+ assert op_id in self._start_times, "No operation with the specified op_id in this schema."
+ return self._start_times[op_id]
+
+ def forward_slack(self, op_id):
+ raise NotImplementedError
+
+ def backward_slack(self, op_id):
+ raise NotImplementedError
+
+ def print_slacks(self):
+ raise NotImplementedError
+
+ def _schedule_asap(self):
+ pl = self._sfg.get_precedence_list()
+
+ if len(pl) < 2:
+ print("Empty signal flow graph cannot be scheduled.")
+ return
+
+ non_schedulable_ops = set((outp.operation.graph_id for outp in pl[0]))
+
+ for outport in pl[1]:
+ op = outport.operation
+ if op not in self._start_times:
+ # Set start time of all operations in the first iter to 0
+ self._start_times[op.graph_id] = 0
+
+ for outports in pl[2:]:
+ for outport in outports:
+ op = outport.operation
+ if op.graph_id not in self._start_times:
+ # Schedule the operation if it doesn't have a start time yet.
+ op_start_time = 0
+ for inport in op.inputs:
+ print(inport.operation.graph_id)
+ assert len(inport.signals) == 1, "Error in scheduling, dangling input port detected."
+ assert inport.signals[0].source is not None, "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:
+ source_end_time = 0
+ else:
+ source_op_time = self._start_times[source_port.operation.graph_id]
+
+ assert source_port.latency_offset is not None, f"Output port: {source_port.index} of operation: \
+ {source_port.operation.graph_id} has no latency-offset."
+ assert inport.latency_offset is not None, f"Input port: {inport.index} of operation: \
+ {inport.operation.graph_id} has no latency-offset."
+
+ source_end_time = source_op_time + source_port.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)
+
+ self._start_times[op.graph_id] = op_start_time
diff --git a/b_asic/signal_flow_graph.py b/b_asic/signal_flow_graph.py
index d51f13b4209fd1d5fc87e369b2f23dc8bf69301b..1a0cd8e7cee0f1ae8798233348a7b45e12f7e74f 100644
--- a/b_asic/signal_flow_graph.py
+++ b/b_asic/signal_flow_graph.py
@@ -8,7 +8,7 @@ from numbers import Number
from collections import defaultdict, deque
from io import StringIO
from queue import PriorityQueue
-import itertools
+import itertools as it
from graphviz import Digraph
from b_asic.port import SignalSourceProvider, OutputPort
@@ -612,7 +612,7 @@ class SFG(AbstractOperation):
def show_precedence_graph(self) -> None:
p_list = self.get_precedence_list()
pg = Digraph()
- pg.attr(rankdir = 'LR')
+ pg.attr(rankdir='LR')
# Creates nodes for each output port in the precedence list
for i in range(len(p_list)):
@@ -627,11 +627,11 @@ class SFG(AbstractOperation):
for port in ports:
for signal in port.signals:
pg.edge(port.operation.graph_id + '.' + str(port.index), signal.destination.operation.graph_id)
- pg.node(signal.destination.operation.graph_id, shape = 'square')
+ pg.node(signal.destination.operation.graph_id, shape='square')
pg.edge(port.operation.graph_id, port.operation.graph_id + '.' + str(port.index))
- pg.node(port.operation.graph_id, shape = 'square')
+ pg.node(port.operation.graph_id, shape='square')
- pg.view()
+ pg.view()
def print_precedence_graph(self) -> None:
"""Prints a representation of the SFG's precedence list to the standard out.
@@ -682,14 +682,18 @@ class SFG(AbstractOperation):
first_op = no_inputs_queue.popleft()
visited = set([first_op])
p_queue = PriorityQueue()
- p_queue.put((-first_op.output_count, first_op)) # Negative priority as max-heap popping is wanted
+ p_queue_entry_num = it.count()
+ # Negative priority as max-heap popping is wanted
+ p_queue.put((-first_op.output_count, -next(p_queue_entry_num), first_op))
+
operations_left = len(self.operations) - 1
seen_but_not_visited_count = 0
while operations_left > 0:
while not p_queue.empty():
- op = p_queue.get()[1]
+
+ op = p_queue.get()[2]
operations_left -= 1
top_order.append(op)
@@ -701,7 +705,7 @@ class SFG(AbstractOperation):
remaining_inports = remaining_inports_per_operation[neighbor_op]
if remaining_inports == 0:
- p_queue.put((-neighbor_op.output_count, neighbor_op))
+ p_queue.put((-neighbor_op.output_count, -next(p_queue_entry_num), neighbor_op))
elif remaining_inports > 0:
if neighbor_op in seen:
@@ -717,15 +721,15 @@ class SFG(AbstractOperation):
# First check if can fetch from Operations with no input ports
if no_inputs_queue:
new_op = no_inputs_queue.popleft()
- p_queue.put((new_op.output_count, new_op))
+ p_queue.put((-new_op.output_count, -next(p_queue_entry_num), new_op))
# Else fetch operation with lowest input count that is not zero
elif seen_but_not_visited_count > 0:
- for i in itertools.count(start=1):
+ for i in it.count(start=1):
seen_inputs_queue = seen_with_inputs_dict[i]
if seen_inputs_queue:
new_op = seen_inputs_queue.popleft()
- p_queue.put((-new_op.output_count, new_op))
+ p_queue.put((-new_op.output_count, -next(p_queue_entry_num), new_op))
seen_but_not_visited_count -= 1
break
else:
@@ -734,3 +738,11 @@ class SFG(AbstractOperation):
self._operations_topological_order = top_order
return self._operations_topological_order
+
+ def set_latency_of_type(self, type_name: TypeName, latency: int):
+ for op in self.get_components_with_type_name(type_name):
+ op.set_latency(latency)
+
+ def set_latency_offsets_of_type(self, type_name: TypeName, latency_offsets: Dict[str, int]):
+ for op in self.get_components_with_type_name(type_name):
+ op.set_latency_offsets(latency_offsets)
diff --git a/b_asic/special_operations.py b/b_asic/special_operations.py
index 0a256bc86c7d5e24582a54d95b7d3afbef1ad941..14440eb098e7100e5399f3ec023b17bcab4ed458 100644
--- a/b_asic/special_operations.py
+++ b/b_asic/special_operations.py
@@ -6,7 +6,7 @@ TODO: More info.
from numbers import Number
from typing import Optional, Sequence
-from b_asic.operation import AbstractOperation, OutputKey, RegisterMap, MutableOutputMap, MutableRegisterMap
+from b_asic.operation import AbstractOperation, RegisterMap, MutableOutputMap, MutableRegisterMap
from b_asic.graph_component import Name, TypeName
from b_asic.port import SignalSourceProvider
diff --git a/test/fixtures/signal_flow_graph.py b/test/fixtures/signal_flow_graph.py
index 08d9e8aa2bacd0b1c1a11c17c174179d853e6ed7..6cf434dbfa5fbad4380e4b4d05ef07b9a02e30ec 100644
--- a/test/fixtures/signal_flow_graph.py
+++ b/test/fixtures/signal_flow_graph.py
@@ -21,12 +21,12 @@ def sfg_two_inputs_two_outputs():
out1 = in1 + in2
out2 = in1 + 2 * in2
"""
- in1 = Input()
- in2 = Input()
- add1 = in1 + in2
- add2 = add1 + in2
- out1 = Output(add1)
- out2 = Output(add2)
+ in1 = Input("IN1")
+ in2 = Input("IN2")
+ add1 = Addition(in1, in2, "ADD1")
+ add2 = Addition(add1, in2, "ADD2")
+ out1 = Output(add1, "OUT1")
+ out2 = Output(add2, "OUT2")
return SFG(inputs=[in1, in2], outputs=[out1, out2])
@@ -58,6 +58,31 @@ def sfg_two_inputs_two_outputs_independent():
return SFG(inputs=[in1, in2], outputs=[out1, out2])
+@pytest.fixture
+def sfg_two_inputs_two_outputs_independent_with_cmul():
+ """Valid SFG with two inputs and two outputs, where the first output only depends
+ on the first input and the second output only depends on the second input.
+ . .
+ in1--->cmul1--->cmul2--->out1
+ . .
+ . .
+ c1------+ .
+ |
+ v .
+ in2--->add1---->cmul3--->out2
+ """
+ in1 = Input("IN1")
+ in2 = Input("IN2")
+ c1 = Constant(3, "C1")
+ add1 = Addition(in2, c1, "ADD1", 7)
+ cmul3 = ConstantMultiplication(2, add1, "CMUL3", 3)
+ cmul1 = ConstantMultiplication(5, in1, "CMUL1", 5)
+ cmul2 = ConstantMultiplication(4, cmul1, "CMUL2", 4)
+ out1 = Output(in1, "OUT1")
+ out2 = Output(add1, "OUT2")
+ return SFG(inputs=[in1, in2], outputs=[out1, out2])
+
+
@pytest.fixture
def sfg_nested():
"""Valid SFG with two inputs and one output.
diff --git a/test/test_abstract_operation.py b/test/test_abstract_operation.py
deleted file mode 100644
index 9163fce2a955c7fbc68d5d24de86896d251934da..0000000000000000000000000000000000000000
--- a/test/test_abstract_operation.py
+++ /dev/null
@@ -1,91 +0,0 @@
-"""
-B-ASIC test suite for the AbstractOperation class.
-"""
-
-import pytest
-
-from b_asic import Addition, Subtraction, Multiplication, ConstantMultiplication, Division
-
-
-def test_addition_overload():
- """Tests addition overloading for both operation and number argument."""
- add1 = Addition(None, None, "add1")
- add2 = Addition(None, None, "add2")
-
- add3 = add1 + add2
- assert isinstance(add3, Addition)
- assert add3.input(0).signals == add1.output(0).signals
- assert add3.input(1).signals == add2.output(0).signals
-
- add4 = add3 + 5
- assert isinstance(add4, Addition)
- assert add4.input(0).signals == add3.output(0).signals
- assert add4.input(1).signals[0].source.operation.value == 5
-
- add5 = 5 + add4
- assert isinstance(add5, Addition)
- assert add5.input(0).signals[0].source.operation.value == 5
- assert add5.input(1).signals == add4.output(0).signals
-
-
-def test_subtraction_overload():
- """Tests subtraction overloading for both operation and number argument."""
- add1 = Addition(None, None, "add1")
- add2 = Addition(None, None, "add2")
-
- sub1 = add1 - add2
- assert isinstance(sub1, Subtraction)
- assert sub1.input(0).signals == add1.output(0).signals
- assert sub1.input(1).signals == add2.output(0).signals
-
- sub2 = sub1 - 5
- assert isinstance(sub2, Subtraction)
- assert sub2.input(0).signals == sub1.output(0).signals
- assert sub2.input(1).signals[0].source.operation.value == 5
-
- sub3 = 5 - sub2
- assert isinstance(sub3, Subtraction)
- assert sub3.input(0).signals[0].source.operation.value == 5
- assert sub3.input(1).signals == sub2.output(0).signals
-
-
-def test_multiplication_overload():
- """Tests multiplication overloading for both operation and number argument."""
- add1 = Addition(None, None, "add1")
- add2 = Addition(None, None, "add2")
-
- mul1 = add1 * add2
- assert isinstance(mul1, Multiplication)
- assert mul1.input(0).signals == add1.output(0).signals
- assert mul1.input(1).signals == add2.output(0).signals
-
- mul2 = mul1 * 5
- assert isinstance(mul2, ConstantMultiplication)
- assert mul2.input(0).signals == mul1.output(0).signals
- assert mul2.value == 5
-
- mul3 = 5 * mul2
- assert isinstance(mul3, ConstantMultiplication)
- assert mul3.input(0).signals == mul2.output(0).signals
- assert mul3.value == 5
-
-
-def test_division_overload():
- """Tests division overloading for both operation and number argument."""
- add1 = Addition(None, None, "add1")
- add2 = Addition(None, None, "add2")
-
- div1 = add1 / add2
- assert isinstance(div1, Division)
- assert div1.input(0).signals == add1.output(0).signals
- assert div1.input(1).signals == add2.output(0).signals
-
- div2 = div1 / 5
- assert isinstance(div2, Division)
- assert div2.input(0).signals == div1.output(0).signals
- assert div2.input(1).signals[0].source.operation.value == 5
-
- div3 = 5 / div2
- assert isinstance(div3, Division)
- assert div3.input(0).signals[0].source.operation.value == 5
- assert div3.input(1).signals == div2.output(0).signals
diff --git a/test/test_operation.py b/test/test_operation.py
index 77e9ba3cbd0eaa75886b5a7e5d11f00f6cfeb479..f4af81b57b8d30fe71025ab82f75f57f195ce350 100644
--- a/test/test_operation.py
+++ b/test/test_operation.py
@@ -1,6 +1,94 @@
+"""
+B-ASIC test suite for the AbstractOperation class.
+"""
+
import pytest
-from b_asic import Constant, Addition, MAD, Butterfly, SquareRoot
+from b_asic import Addition, Subtraction, Multiplication, ConstantMultiplication, Division, Constant, Butterfly, \
+ MAD, SquareRoot
+
+
+class TestOperationOverloading:
+ def test_addition_overload(self):
+ """Tests addition overloading for both operation and number argument."""
+ add1 = Addition(None, None, "add1")
+ add2 = Addition(None, None, "add2")
+
+ add3 = add1 + add2
+ assert isinstance(add3, Addition)
+ assert add3.input(0).signals == add1.output(0).signals
+ assert add3.input(1).signals == add2.output(0).signals
+
+ add4 = add3 + 5
+ assert isinstance(add4, Addition)
+ assert add4.input(0).signals == add3.output(0).signals
+ assert add4.input(1).signals[0].source.operation.value == 5
+
+ add5 = 5 + add4
+ assert isinstance(add5, Addition)
+ assert add5.input(0).signals[0].source.operation.value == 5
+ assert add5.input(1).signals == add4.output(0).signals
+
+ def test_subtraction_overload(self):
+ """Tests subtraction overloading for both operation and number argument."""
+ add1 = Addition(None, None, "add1")
+ add2 = Addition(None, None, "add2")
+
+ sub1 = add1 - add2
+ assert isinstance(sub1, Subtraction)
+ assert sub1.input(0).signals == add1.output(0).signals
+ assert sub1.input(1).signals == add2.output(0).signals
+
+ sub2 = sub1 - 5
+ assert isinstance(sub2, Subtraction)
+ assert sub2.input(0).signals == sub1.output(0).signals
+ assert sub2.input(1).signals[0].source.operation.value == 5
+
+ sub3 = 5 - sub2
+ assert isinstance(sub3, Subtraction)
+ assert sub3.input(0).signals[0].source.operation.value == 5
+ assert sub3.input(1).signals == sub2.output(0).signals
+
+ def test_multiplication_overload(self):
+ """Tests multiplication overloading for both operation and number argument."""
+ add1 = Addition(None, None, "add1")
+ add2 = Addition(None, None, "add2")
+
+ mul1 = add1 * add2
+ assert isinstance(mul1, Multiplication)
+ assert mul1.input(0).signals == add1.output(0).signals
+ assert mul1.input(1).signals == add2.output(0).signals
+
+ mul2 = mul1 * 5
+ assert isinstance(mul2, ConstantMultiplication)
+ assert mul2.input(0).signals == mul1.output(0).signals
+ assert mul2.value == 5
+
+ mul3 = 5 * mul2
+ assert isinstance(mul3, ConstantMultiplication)
+ assert mul3.input(0).signals == mul2.output(0).signals
+ assert mul3.value == 5
+
+ def test_division_overload(self):
+ """Tests division overloading for both operation and number argument."""
+ add1 = Addition(None, None, "add1")
+ add2 = Addition(None, None, "add2")
+
+ div1 = add1 / add2
+ assert isinstance(div1, Division)
+ assert div1.input(0).signals == add1.output(0).signals
+ assert div1.input(1).signals == add2.output(0).signals
+
+ div2 = div1 / 5
+ assert isinstance(div2, Division)
+ assert div2.input(0).signals == div1.output(0).signals
+ assert div2.input(1).signals[0].source.operation.value == 5
+
+ div3 = 5 / div2
+ assert isinstance(div3, Division)
+ assert div3.input(0).signals[0].source.operation.value == 5
+ assert div3.input(1).signals == div2.output(0).signals
+
class TestTraverse:
def test_traverse_single_tree(self, operation):
@@ -24,20 +112,21 @@ class TestTraverse:
def test_traverse_loop(self, operation_graph_with_cycle):
assert len(list(operation_graph_with_cycle.traverse())) == 8
+
class TestToSfg:
def test_convert_mad_to_sfg(self):
mad1 = MAD()
mad1_sfg = mad1.to_sfg()
- assert mad1.evaluate(1,1,1) == mad1_sfg.evaluate(1,1,1)
+ assert mad1.evaluate(1, 1, 1) == mad1_sfg.evaluate(1, 1, 1)
assert len(mad1_sfg.operations) == 6
def test_butterfly_to_sfg(self):
but1 = Butterfly()
but1_sfg = but1.to_sfg()
- assert but1.evaluate(1,1)[0] == but1_sfg.evaluate(1,1)[0]
- assert but1.evaluate(1,1)[1] == but1_sfg.evaluate(1,1)[1]
+ assert but1.evaluate(1, 1)[0] == but1_sfg.evaluate(1, 1)[0]
+ assert but1.evaluate(1, 1)[1] == but1_sfg.evaluate(1, 1)[1]
assert len(but1_sfg.operations) == 8
def test_add_to_sfg(self):
@@ -51,3 +140,47 @@ class TestToSfg:
sqrt1_sfg = sqrt1.to_sfg()
assert len(sqrt1_sfg.operations) == 3
+
+
+class TestLatency:
+ def test_latency_constructor(self):
+ bfly = Butterfly(latency=5)
+
+ assert bfly.latency == 5
+ assert bfly.latency_offsets == {'in0': 0, 'in1': 0, 'out0': 5, 'out1': 5}
+
+ def test_latency_offsets_constructor(self):
+ bfly = Butterfly(latency_offsets={'in0': 2, 'in1': 3, 'out0': 5, 'out1': 10})
+
+ assert bfly.latency == 8
+ assert bfly.latency_offsets == {'in0': 2, 'in1': 3, 'out0': 5, 'out1': 10}
+
+ def test_latency_and_latency_offsets_constructor(self):
+ bfly = Butterfly(latency=5, latency_offsets={'in1': 2, 'out0': 9})
+
+ assert bfly.latency == 9
+ assert bfly.latency_offsets == {"in0": 0, "in1": 2, "out0": 9, "out1": 5}
+
+ def test_set_latency(self):
+ bfly = Butterfly()
+
+ bfly.set_latency(9)
+
+ assert bfly.latency == 9
+ assert bfly.latency_offsets == {"in0": 0, "in1": 0, "out0": 9, "out1": 9}
+
+ def test_set_latency_offsets(self):
+ bfly = Butterfly()
+
+ bfly.set_latency_offsets({'in0': 3, 'out1': 5})
+
+ assert bfly.latency_offsets == {'in0': 3, "in1": None, "out0": None, 'out1': 5}
+
+
+class TestCopyOperation:
+ def test_copy_buttefly_latency_offsets(self):
+ bfly = Butterfly(latency_offsets={'in0': 4, 'in1': 2, 'out0': 10, 'out1': 9})
+
+ bfly_copy = bfly.copy_component()
+
+ assert bfly_copy.latency_offsets == {'in0': 4, 'in1': 2, 'out0': 10, 'out1': 9}
diff --git a/test/test_schema.py b/test/test_schema.py
new file mode 100644
index 0000000000000000000000000000000000000000..510a0d997cb6cbdb525d600f992cbbae3468e8dd
--- /dev/null
+++ b/test/test_schema.py
@@ -0,0 +1,67 @@
+"""
+B-ASIC test suite for the schema module and Schema class.
+"""
+
+from b_asic import Schema, Addition, ConstantMultiplication
+
+
+class TestInit:
+ def test_simple_filter_normal_latency(self, simple_filter):
+ simple_filter.set_latency_of_type(Addition.type_name(), 5)
+ simple_filter.set_latency_of_type(ConstantMultiplication.type_name(), 4)
+
+ schema = Schema(simple_filter)
+
+ assert schema._start_times == {"add1": 4, "cmul1": 0}
+
+ def test_complicated_single_outputs_normal_latency(self, precedence_sfg_registers):
+ precedence_sfg_registers.set_latency_of_type(Addition.type_name(), 4)
+ precedence_sfg_registers.set_latency_of_type(ConstantMultiplication.type_name(), 3)
+
+ schema = Schema(precedence_sfg_registers, scheduling_alg="ASAP")
+
+ for op in schema._sfg.get_operations_topological_order():
+ print(op.latency_offsets)
+
+ start_times_names = dict()
+ for op_id, start_time in schema._start_times.items():
+ op_name = precedence_sfg_registers.find_by_id(op_id).name
+ start_times_names[op_name] = start_time
+
+ assert start_times_names == {"C0": 0, "B1": 0, "B2": 0, "ADD2": 3, "ADD1": 7, "Q1": 11,
+ "A0": 14, "A1": 0, "A2": 0, "ADD3": 3, "ADD4": 17}
+
+ def test_complicated_single_outputs_complex_latencies(self, precedence_sfg_registers):
+ precedence_sfg_registers.set_latency_offsets_of_type(ConstantMultiplication.type_name(), {'in0': 3, 'out0': 5})
+
+ precedence_sfg_registers.find_by_name("B1")[0].set_latency_offsets({'in0': 4, 'out0': 7})
+ precedence_sfg_registers.find_by_name("B2")[0].set_latency_offsets({'in0': 1, 'out0': 4})
+ precedence_sfg_registers.find_by_name("ADD2")[0].set_latency_offsets({'in0': 4, 'in1': 2, 'out0': 4})
+ precedence_sfg_registers.find_by_name("ADD1")[0].set_latency_offsets({'in0': 1, 'in1': 2, 'out0': 4})
+ precedence_sfg_registers.find_by_name("Q1")[0].set_latency_offsets({'in0': 3, 'out0': 6})
+ precedence_sfg_registers.find_by_name("A0")[0].set_latency_offsets({'in0': 0, 'out0': 2})
+
+ precedence_sfg_registers.find_by_name("A1")[0].set_latency_offsets({'in0': 0, 'out0': 5})
+ precedence_sfg_registers.find_by_name("A2")[0].set_latency_offsets({'in0': 2, 'out0': 3})
+ precedence_sfg_registers.find_by_name("ADD3")[0].set_latency_offsets({'in0': 2, 'in1': 1, 'out0': 4})
+ precedence_sfg_registers.find_by_name("ADD4")[0].set_latency_offsets({'in0': 6, 'in1': 7, 'out0': 9})
+
+ schema = Schema(precedence_sfg_registers, scheduling_alg="ASAP")
+
+ start_times_names = dict()
+ for op_id, start_time in schema._start_times.items():
+ op_name = precedence_sfg_registers.find_by_id(op_id).name
+ start_times_names[op_name] = start_time
+
+ assert start_times_names == {'C0': 0, 'B1': 0, 'B2': 0, 'ADD2': 3, 'ADD1': 5, 'Q1': 6, 'A0': 12,
+ 'A1': 0, 'A2': 0, 'ADD3': 3, 'ADD4': 8}
+
+ def test_independent_sfg(self, sfg_two_inputs_two_outputs_independent_with_cmul):
+ schema = Schema(sfg_two_inputs_two_outputs_independent_with_cmul, scheduling_alg="ASAP")
+
+ start_times_names = dict()
+ for op_id, start_time in schema._start_times.items():
+ 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 == {'CMUL1': 0, 'CMUL2': 5, "ADD1": 0, "CMUL3": 7}
diff --git a/test/test_sfg.py b/test/test_sfg.py
index 241a38b34c2177ffda9dbde88299178591644281..1dfca903d387693f7ba8edf872e292717f056c94 100644
--- a/test/test_sfg.py
+++ b/test/test_sfg.py
@@ -702,6 +702,7 @@ class TestConnectExternalSignalsToComponentsMultipleComp:
assert test_sfg.evaluate(1, 2, 3, 4) == 16
assert not test_sfg.connect_external_signals_to_components()
+
class TestTopologicalOrderOperations:
def test_feedback_sfg(self, simple_filter):
topological_order = simple_filter.get_operations_topological_order()
@@ -713,6 +714,12 @@ class TestTopologicalOrderOperations:
assert [comp.name for comp in topological_order] == ["IN1", "OUT1", "IN2", "C1", "ADD1", "OUT2"]
+ def test_complex_graph(self, precedence_sfg_registers):
+ topological_order = precedence_sfg_registers.get_operations_topological_order()
+
+ assert [comp.name for comp in topological_order] == \
+ ['IN1', 'C0', 'ADD1', 'Q1', 'A0', 'T1', 'B1', 'A1', 'T2', 'B2', 'ADD2', 'A2', 'ADD3', 'ADD4', 'OUT1']
+
class TestRemove:
def test_remove_single_input_outputs(self, simple_filter):
@@ -851,7 +858,6 @@ class TestSaveLoadSFG:
precedence_sfg_registers_and_constants_, _ = python_to_sfg(path_)
assert str(precedence_sfg_registers_and_constants) == str(precedence_sfg_registers_and_constants_)
- assert precedence_sfg_registers_and_constants.evaluate([2]) == precedence_sfg_registers_and_constants_.evaluate([2])
remove(path_)
@@ -859,3 +865,19 @@ class TestSaveLoadSFG:
path_ = self.get_path(existing=False)
with pytest.raises(Exception):
python_to_sfg(path_)
+
+
+class TestGetComponentsOfType:
+ def test_get_no_operations_of_type(self, sfg_two_inputs_two_outputs):
+ assert [op.name for op in sfg_two_inputs_two_outputs.get_components_with_type_name(Multiplication.type_name())] \
+ == []
+
+ def test_get_multple_operations_of_type(self, sfg_two_inputs_two_outputs):
+ assert [op.name for op in sfg_two_inputs_two_outputs.get_components_with_type_name(Addition.type_name())] \
+ == ["ADD1", "ADD2"]
+
+ assert [op.name for op in sfg_two_inputs_two_outputs.get_components_with_type_name(Input.type_name())] \
+ == ["IN1", "IN2"]
+
+ assert [op.name for op in sfg_two_inputs_two_outputs.get_components_with_type_name(Output.type_name())] \
+ == ["OUT1", "OUT2"]