b_asic/GUI/port_button.py

+
+import sys
+
+from PySide2.QtWidgets import QPushButton, QMenu
+from PySide2.QtCore import Qt, Signal
+
+class PortButton(QPushButton):
+    connectionRequested = Signal(QPushButton)
+    moved = Signal()
+    def __init__(self, name, operation, port, window, parent=None):
+        super(PortButton, self).__init__(name, operation, parent)
+        self.pressed = False
+        self._window = window
+        self.port = port
+        self.operation = operation
+        self.clicked = 0
+        self._m_drag = False
+        self._m_press = False
+
+        self.setStyleSheet("background-color: white")
+        self.connectionRequested.connect(self._window.connectButton)
+
+    def contextMenuEvent(self, event):
+        menu = QMenu()
+        menu.addAction("Connect", lambda: self.connectionRequested.emit(self))
+        menu.exec_(self.cursor().pos())
+
+    def mousePressEvent(self, event):
+        if event.button() == Qt.LeftButton:
+            self.select_port(event.modifiers())
+
+        super(PortButton, self).mousePressEvent(event)
+
+    def mouseReleaseEvent(self, event):
+        super(PortButton, self).mouseReleaseEvent(event)
+
+    def _toggle_port(self, pressed=False):
+        self.pressed = not pressed
+        self.setStyleSheet(f"background-color: {'white' if not self.pressed else 'grey'}")
+
+    def select_port(self, modifiers=None):
+        if modifiers != Qt.ControlModifier:
+            for port in self._window.pressed_ports:
+                port._toggle_port(port.pressed)
+
+            self._toggle_port(self.pressed)
+            self._window.pressed_ports = [self]
+
+        else:
+            self._toggle_port(self.pressed)
+            if self in self._window.pressed_ports:
+                self._window.pressed_ports.remove(self)
+            else:
+                self._window.pressed_ports.append(self)
+
+        for signal in self._window.signalList:
+            signal.update()
+

b_asic/GUI/properties_window.py

+from PySide2.QtWidgets import QDialog, QLineEdit, QPushButton, QVBoxLayout, QHBoxLayout,\
+QLabel, QCheckBox
+from PySide2.QtCore import Qt
+from PySide2.QtGui import QIntValidator
+
+class PropertiesWindow(QDialog):
+    def __init__(self, operation, main_window):
+        super(PropertiesWindow, self).__init__()
+        self.operation = operation
+        self._window = main_window
+        self.setWindowFlags(Qt.WindowTitleHint | Qt.WindowCloseButtonHint)
+        self.setWindowTitle("Properties")
+
+        self.name_layout = QHBoxLayout()
+        self.name_layout.setSpacing(50)
+        self.name_label = QLabel("Name:")
+        self.edit_name = QLineEdit(self.operation.operation_path_name)
+        self.name_layout.addWidget(self.name_label)
+        self.name_layout.addWidget(self.edit_name)
+
+        self.vertical_layout = QVBoxLayout()
+        self.vertical_layout.addLayout(self.name_layout)
+
+        if self.operation.operation_path_name == "c":
+            self.constant_layout = QHBoxLayout()
+            self.constant_layout.setSpacing(50)
+            self.constant_value = QLabel("Constant:")
+            self.edit_constant = QLineEdit(str(self.operation.operation.value))
+            self.only_accept_int = QIntValidator()
+            self.edit_constant.setValidator(self.only_accept_int)
+            self.constant_layout.addWidget(self.constant_value)
+            self.constant_layout.addWidget(self.edit_constant)
+            self.vertical_layout.addLayout(self.constant_layout)
+
+        self.show_name_layout = QHBoxLayout()
+        self.check_show_name = QCheckBox("Show name?")
+        if self.operation.is_show_name:
+            self.check_show_name.setChecked(1)
+        else:
+            self.check_show_name.setChecked(0)
+        self.check_show_name.setLayoutDirection(Qt.RightToLeft)
+        self.check_show_name.setStyleSheet("spacing: 170px")
+        self.show_name_layout.addWidget(self.check_show_name)
+        self.vertical_layout.addLayout(self.show_name_layout)
+
+        self.ok = QPushButton("OK")
+        self.ok.clicked.connect(self.save_properties)
+        self.vertical_layout.addWidget(self.ok)
+        self.setLayout(self.vertical_layout)
+
+    def save_properties(self):
+        self._window.logger.info(f"Saving properties of operation: {self.operation.name}.")
+        self.operation.name = self.edit_name.text()
+        self.operation.label.setPlainText(self.operation.name)
+        if self.operation.operation_path_name == "c":
+            self.operation.operation.value = self.edit_constant.text()
+        if self.check_show_name.isChecked():
+            self.operation.label.setOpacity(1)
+            self.operation.is_show_name = True
+        else:
+            self.operation.label.setOpacity(0)
+            self.operation.is_show_name = False
+        self.reject()
\ No newline at end of file

b_asic/GUI/show_pc_window.py

+from PySide2.QtWidgets import QDialog, QPushButton, QVBoxLayout, QCheckBox,\
+QFrame, QFormLayout
+from PySide2.QtCore import Qt, Signal
+
+from b_asic import SFG
+
+class ShowPCWindow(QDialog):
+    pc = Signal()
+
+    def __init__(self, window):
+        super(ShowPCWindow, self).__init__()
+        self._window = window
+        self.check_box_list = []
+        self.setWindowFlags(Qt.WindowTitleHint | Qt.WindowCloseButtonHint)
+        self.setWindowTitle("Show PC")
+
+        self.dialog_layout = QVBoxLayout()
+        self.pc_btn = QPushButton("Show PC")
+        self.pc_btn.clicked.connect(self.show_precedence_graph)
+        self.dialog_layout.addWidget(self.pc_btn)
+        self.setLayout(self.dialog_layout)
+
+    def add_sfg_to_dialog(self):
+        sfg_layout = QVBoxLayout()
+        options_layout = QFormLayout()
+
+        for sfg in self._window.sfg_list:
+            check_box = QCheckBox()
+            options_layout.addRow(sfg.name, check_box)
+            self.check_box_list.append(check_box)
+
+        sfg_layout.addLayout(options_layout)
+
+        frame = QFrame()
+        frame.setFrameShape(QFrame.HLine)
+        frame.setFrameShadow(QFrame.Sunken)
+        self.dialog_layout.addWidget(frame)
+
+        self.dialog_layout.addLayout(sfg_layout)
+
+    def show_precedence_graph(self):
+        for i, check_box in enumerate(self.check_box_list):
+            if check_box.isChecked():
+                self._window.logger.info("Creating a precedence chart from " + self._window.sfg_list[i].name)
+                self._window.sfg_list[i].show_precedence_graph()
+                break
+        
+        self.accept()
+        self.pc.emit()
\ No newline at end of file

b_asic/GUI/simulate_sfg_window.py

+from PySide2.QtWidgets import QDialog, QLineEdit, QPushButton, QVBoxLayout, QHBoxLayout,\
+QLabel, QCheckBox, QSpinBox, QGroupBox, QFrame, QFormLayout, QGridLayout, QSizePolicy, QFileDialog, QShortcut
+from PySide2.QtCore import Qt, Signal
+from PySide2.QtGui import QIntValidator, QKeySequence
+
+from matplotlib.backends import qt_compat
+from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas
+from matplotlib.figure import Figure
+
+
+class SimulateSFGWindow(QDialog):
+    simulate = Signal()
+
+    def __init__(self, window):
+        super(SimulateSFGWindow, self).__init__()
+        self._window = window
+        self.properties = dict()
+        self.sfg_to_layout = dict()
+        self.setWindowFlags(Qt.WindowTitleHint | Qt.WindowCloseButtonHint)
+        self.setWindowTitle("Simulate SFG")
+
+        self.dialog_layout = QVBoxLayout()
+        self.simulate_btn = QPushButton("Simulate")
+        self.simulate_btn.clicked.connect(self.save_properties)
+        self.dialog_layout.addWidget(self.simulate_btn)
+        self.setLayout(self.dialog_layout)
+
+    def add_sfg_to_dialog(self, sfg):
+        sfg_layout = QVBoxLayout()
+        options_layout = QFormLayout()
+
+        name_label = QLabel(f"{sfg.name}")
+        sfg_layout.addWidget(name_label)
+
+        spin_box = QSpinBox()
+        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)
+        sfg_layout.addLayout(options_layout)
+        sfg_layout.addLayout(input_layout)
+
+        frame = QFrame()
+        frame.setFrameShape(QFrame.HLine)
+        frame.setFrameShadow(QFrame.Sunken)
+        self.dialog_layout.addWidget(frame)
+
+        self.sfg_to_layout[sfg] = sfg_layout
+        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]
+            self.properties[sfg] = {
+                "iteration_count": spin_box.value(),
+                "show_plot": check_box_plot.isChecked(),
+                "all_results": check_box_all.isChecked(),
+                "input_values": 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():
+                self.properties[sfg]["all_results"] = True
+
+        self.accept()
+        self.simulate.emit()
+
+
+class Plot(FigureCanvas):
+    def __init__(self, simulation, sfg, window, parent=None, width=5, height=4, dpi=100):
+        self.simulation = simulation
+        self.sfg = sfg
+        self.dpi = dpi
+        self._window = window
+
+        fig = Figure(figsize=(width, height), dpi=dpi)
+        fig.suptitle(sfg.name, fontsize=20)
+        self.axes = fig.add_subplot(111)
+
+        FigureCanvas.__init__(self, fig)
+        self.setParent(parent)
+
+        FigureCanvas.setSizePolicy(self, QSizePolicy.Expanding, QSizePolicy.Expanding)
+        FigureCanvas.updateGeometry(self)
+        self.save_figure = QShortcut(QKeySequence("Ctrl+S"), self)
+        self.save_figure.activated.connect(self._save_plot_figure)
+        self._plot_values_sfg()
+
+    def _save_plot_figure(self):
+        self._window.logger.info(f"Saving plot of figure: {self.sfg.name}.")
+        file_choices = "PNG (*.png)|*.png"
+        path, ext = QFileDialog.getSaveFileName(self, "Save file", "", file_choices)
+        path = path.encode("utf-8")
+        if not path[-4:] == file_choices[-4:].encode("utf-8"):
+            path += file_choices[-4:].encode("utf-8")
+
+        if path:
+            self.print_figure(path.decode(), dpi=self.dpi)
+            self._window.logger.info(f"Saved plot: {self.sfg.name} to path: {path}.")
+
+    def _plot_values_sfg(self):
+        x_axis = list(range(len(self.simulation.results.keys())))
+        for _output in range(self.sfg.output_count):
+            y_axis = list()
+            for _iter in range(len(self.simulation.results.keys())):
+                y_axis.append(self.simulation.results[_iter][str(_output)])
+
+            self.axes.plot(x_axis, y_axis)

b_asic/GUI/utils.py

+from PySide2.QtWidgets import QErrorMessage
+from traceback import format_exc
+
+def handle_error(fn):
+    def wrapper(self, *args, **kwargs):
+        try:
+            return fn(self, *args, **kwargs)
+        except Exception as e:
+            self._window.logger.error(f"Unexpected error: {format_exc()}")
+            QErrorMessage(self._window).showMessage(f"Unexpected error: {format_exc()}")
+
+    return wrapper
+
+def decorate_class(decorator):
+    def decorate(cls):
+        for attr in cls.__dict__:
+            if callable(getattr(cls, attr)):
+                setattr(cls, attr, decorator(getattr(cls, attr)))
+        return cls
+    return decorate
\ No newline at end of file

b_asic/__init__.py

@@ -4,11 +4,10 @@ TODO: More info.
 """
 from b_asic.core_operations import *
 from b_asic.graph_component import *
-from b_asic.graph_id import *
 from b_asic.operation import *
-from b_asic.precedence_chart import *
 from b_asic.port import *
-from b_asic.schema import *
 from b_asic.signal_flow_graph import *
 from b_asic.signal import *
 from b_asic.simulation import *
+from b_asic.special_operations import *
+from b_asic.schema import *

b_asic/core_operations.py

@@ -4,43 +4,39 @@ TODO: More info.
 """
 
 from numbers import Number
-from typing import Any
+from typing import Optional, Dict
 from numpy import conjugate, sqrt, abs as np_abs
-from b_asic.port import InputPort, OutputPort
-from b_asic.graph_id import GraphIDType
+
+from b_asic.port import SignalSourceProvider, InputPort, OutputPort
 from b_asic.operation import AbstractOperation
 from b_asic.graph_component import Name, TypeName
 
 
-class Input(AbstractOperation):
-    """Input operation.
-    TODO: More info.
-    """
-
-    # TODO: Implement all functions.
-
-    @property
-    def type_name(self) -> TypeName:
-        return "in"
-
-
 class Constant(AbstractOperation):
     """Constant value operation.
     TODO: More info.
     """
 
     def __init__(self, value: Number = 0, name: Name = ""):
-        super().__init__(name)
+        super().__init__(input_count=0, output_count=1, name=name)
+        self.set_param("value", value)
 
-        self._output_ports = [OutputPort(0, self)]
-        self._parameters["value"] = value
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "c"
 
     def evaluate(self):
         return self.param("value")
 
     @property
-    def type_name(self) -> TypeName:
-        return "c"
+    def value(self) -> Number:
+        """Get the constant value of this operation."""
+        return self.param("value")
+
+    @value.setter
+    def value(self, value: Number) -> None:
+        """Set the constant value of this operation."""
+        return self.set_param("value", value)
 
 
 class Addition(AbstractOperation):
@@ -48,134 +44,86 @@ class Addition(AbstractOperation):
     TODO: More info.
     """
 
-    def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
-        super().__init__(name)
-
-        self._input_ports = [InputPort(0, self), InputPort(1, self)]
-        self._output_ports = [OutputPort(0, self)]
+    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)
 
-        if source1 is not None:
-            self._input_ports[0].connect(source1)
-        if source2 is not None:
-            self._input_ports[1].connect(source2)
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "add"
 
     def evaluate(self, a, b):
         return a + b
 
-    @property
-    def type_name(self) -> TypeName:
-        return "add"
-
 
 class Subtraction(AbstractOperation):
     """Binary subtraction operation.
     TODO: More info.
     """
 
-    def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
-        super().__init__(name)
-        self._input_ports = [InputPort(0, self), InputPort(1, self)]
-        self._output_ports = [OutputPort(0, self)]
+    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)
 
-        if source1 is not None:
-            self._input_ports[0].connect(source1)
-        if source2 is not None:
-            self._input_ports[1].connect(source2)
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "sub"
 
     def evaluate(self, a, b):
         return a - b
 
-    @property
-    def type_name(self) -> TypeName:
-        return "sub"
-
 
 class Multiplication(AbstractOperation):
     """Binary multiplication operation.
     TODO: More info.
     """
 
-    def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
-        super().__init__(name)
-        self._input_ports = [InputPort(0, self), InputPort(1, self)]
-        self._output_ports = [OutputPort(0, self)]
+    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)
 
-        if source1 is not None:
-            self._input_ports[0].connect(source1)
-        if source2 is not None:
-            self._input_ports[1].connect(source2)
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "mul"
 
     def evaluate(self, a, b):
         return a * b
 
-    @property
-    def type_name(self) -> TypeName:
-        return "mul"
-
 
 class Division(AbstractOperation):
     """Binary division operation.
     TODO: More info.
     """
 
-    def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
-        super().__init__(name)
-        self._input_ports = [InputPort(0, self), InputPort(1, self)]
-        self._output_ports = [OutputPort(0, self)]
+    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)
 
-        if source1 is not None:
-            self._input_ports[0].connect(source1)
-        if source2 is not None:
-            self._input_ports[1].connect(source2)
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "div"
 
     def evaluate(self, a, b):
         return a / b
 
-    @property
-    def type_name(self) -> TypeName:
-        return "div"
-
-
-class SquareRoot(AbstractOperation):
-    """Unary square root operation.
-    TODO: More info.
-    """
-
-    def __init__(self, source1: OutputPort = None, name: Name = ""):
-        super().__init__(name)
-        self._input_ports = [InputPort(0, self)]
-        self._output_ports = [OutputPort(0, self)]
-
-        if source1 is not None:
-            self._input_ports[0].connect(source1)
-
-    def evaluate(self, a):
-        return sqrt((complex)(a))
-
-    @property
-    def type_name(self) -> TypeName:
-        return "sqrt"
 
-
-class ComplexConjugate(AbstractOperation):
-    """Unary complex conjugate operation.
+class Min(AbstractOperation):
+    """Binary min operation.
     TODO: More info.
     """
 
-    def __init__(self, source1: OutputPort = None, name: Name = ""):
-        super().__init__(name)
-        self._input_ports = [InputPort(0, self)]
-        self._output_ports = [OutputPort(0, self)]
+    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)
 
-        if source1 is not None:
-            self._input_ports[0].connect(source1)
-
-    def evaluate(self, a):
-        return conjugate(a)
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "min"
 
-    @property
-    def type_name(self) -> TypeName:
-        return "conj"
+    def evaluate(self, a, b):
+        assert not isinstance(a, complex) and not isinstance(b, complex), \
+            ("core_operations.Min does not support complex numbers.")
+        return a if a < b else b
 
 
 class Max(AbstractOperation):
@@ -183,155 +131,129 @@ class Max(AbstractOperation):
     TODO: More info.
     """
 
-    def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
-        super().__init__(name)
-        self._input_ports = [InputPort(0, self), InputPort(1, self)]
-        self._output_ports = [OutputPort(0, self)]
+    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)
 
-        if source1 is not None:
-            self._input_ports[0].connect(source1)
-        if source2 is not None:
-            self._input_ports[1].connect(source2)
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "max"
 
     def evaluate(self, a, b):
         assert not isinstance(a, complex) and not isinstance(b, complex), \
             ("core_operations.Max does not support complex numbers.")
         return a if a > b else b
 
-    @property
-    def type_name(self) -> TypeName:
-        return "max"
 
-
-class Min(AbstractOperation):
-    """Binary min operation.
+class SquareRoot(AbstractOperation):
+    """Unary square root operation.
     TODO: More info.
     """
 
-    def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
-        super().__init__(name)
-        self._input_ports = [InputPort(0, self), InputPort(1, self)]
-        self._output_ports = [OutputPort(0, self)]
-
-        if source1 is not None:
-            self._input_ports[0].connect(source1)
-        if source2 is not None:
-            self._input_ports[1].connect(source2)
+    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)
 
-    def evaluate(self, a, b):
-        assert not isinstance(a, complex) and not isinstance(b, complex), \
-            ("core_operations.Min does not support complex numbers.")
-        return a if a < b else b
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "sqrt"
 
-    @property
-    def type_name(self) -> TypeName:
-        return "min"
+    def evaluate(self, a):
+        return sqrt(complex(a))
 
 
-class Absolute(AbstractOperation):
-    """Unary absolute value operation.
+class ComplexConjugate(AbstractOperation):
+    """Unary complex conjugate operation.
     TODO: More info.
     """
 
-    def __init__(self, source1: OutputPort = None, name: Name = ""):
-        super().__init__(name)
-        self._input_ports = [InputPort(0, self)]
-        self._output_ports = [OutputPort(0, self)]
+    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)
 
-        if source1 is not None:
-            self._input_ports[0].connect(source1)
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "conj"
 
     def evaluate(self, a):
-        return np_abs(a)
-
-    @property
-    def type_name(self) -> TypeName:
-        return "abs"
+        return conjugate(a)
 
 
-class ConstantMultiplication(AbstractOperation):
-    """Unary constant multiplication operation.
+class Absolute(AbstractOperation):
+    """Unary absolute value operation.
     TODO: More info.
     """
 
-    def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
-        super().__init__(name)
-        self._input_ports = [InputPort(0, self)]
-        self._output_ports = [OutputPort(0, self)]
-        self._parameters["coefficient"] = coefficient
+    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)
 
-        if source1 is not None:
-            self._input_ports[0].connect(source1)
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "abs"
 
     def evaluate(self, a):
-        return a * self.param("coefficient")
-
-    @property
-    def type_name(self) -> TypeName:
-        return "cmul"
+        return np_abs(a)
 
 
-class ConstantAddition(AbstractOperation):
-    """Unary constant addition operation.
+class ConstantMultiplication(AbstractOperation):
+    """Unary constant multiplication operation.
     TODO: More info.
     """
 
-    def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
-        super().__init__(name)
-        self._input_ports = [InputPort(0, self)]
-        self._output_ports = [OutputPort(0, self)]
-        self._parameters["coefficient"] = coefficient
+    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)
 
-        if source1 is not None:
-            self._input_ports[0].connect(source1)
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "cmul"
 
     def evaluate(self, a):
-        return a + self.param("coefficient")
+        return a * self.param("value")
 
     @property
-    def type_name(self) -> TypeName:
-        return "cadd"
+    def value(self) -> Number:
+        """Get the constant value of this operation."""
+        return self.param("value")
+
+    @value.setter
+    def value(self, value: Number) -> None:
+        """Set the constant value of this operation."""
+        return self.set_param("value", value)
 
 
-class ConstantSubtraction(AbstractOperation):
-    """Unary constant subtraction operation.
+class Butterfly(AbstractOperation):
+    """Butterfly operation that returns two outputs.
+    The first output is a + b and the second output is a - b.
     TODO: More info.
     """
 
-    def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
-        super().__init__(name)
-        self._input_ports = [InputPort(0, self)]
-        self._output_ports = [OutputPort(0, self)]
-        self._parameters["coefficient"] = coefficient
+    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)
 
-        if source1 is not None:
-            self._input_ports[0].connect(source1)
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "bfly"
 
-    def evaluate(self, a):
-        return a - self.param("coefficient")
-
-    @property
-    def type_name(self) -> TypeName:
-        return "csub"
+    def evaluate(self, a, b):
+        return a + b, a - b
 
 
-class ConstantDivision(AbstractOperation):
-    """Unary constant division operation.
+class MAD(AbstractOperation):
+    """Multiply-and-add operation.
     TODO: More info.
     """
 
-    def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
-        super().__init__(name)
-        self._input_ports = [InputPort(0, self)]
-        self._output_ports = [OutputPort(0, self)]
-        self._parameters["coefficient"] = coefficient
+    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)
 
-        if source1 is not None:
-            self._input_ports[0].connect(source1)
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "mad"
 
-    def evaluate(self, a):
-        return a / self.param("coefficient")
-
-    @property
-    def type_name(self) -> TypeName:
-        return "cdiv"
+    def evaluate(self, a, b, c):
+        return a * b + c

b_asic/graph_component.py

@@ -4,10 +4,15 @@ TODO: More info.
 """
 
 from abc import ABC, abstractmethod
-from typing import NewType
+from collections import deque
+from copy import copy, deepcopy
+from typing import NewType, Any, Dict, Mapping, Iterable, Generator
+
 
 Name = NewType("Name", str)
 TypeName = NewType("TypeName", str)
+GraphID = NewType("GraphID", str)
+GraphIDNumber = NewType("GraphIDNumber", int)
 
 
 class GraphComponent(ABC):
@@ -15,35 +20,94 @@ class GraphComponent(ABC):
     TODO: More info.
     """
 
-    @property
+    @classmethod
     @abstractmethod
-    def type_name(self) -> TypeName:
-        """Return the type name of the graph component"""
+    def type_name(cls) -> TypeName:
+        """Get the type name of this graph component"""
         raise NotImplementedError
 
     @property
     @abstractmethod
     def name(self) -> Name:
-        """Return the name of the graph component."""
+        """Get the name of this graph component."""
         raise NotImplementedError
 
     @name.setter
     @abstractmethod
     def name(self, name: Name) -> None:
-        """Set the name of the graph component to the entered name."""
+        """Set the name of this graph component to the given name."""
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def graph_id(self) -> GraphID:
+        """Get the graph id of this graph component."""
+        raise NotImplementedError
+
+    @graph_id.setter
+    @abstractmethod
+    def graph_id(self, graph_id: GraphID) -> None:
+        """Set the graph id of this graph component to the given id.
+        Note that this id will be ignored if this component is used to create a new graph,
+        and that a new local id will be generated for it instead."""
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def params(self) -> Mapping[str, Any]:
+        """Get a dictionary of all parameter values."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def param(self, name: str) -> Any:
+        """Get the value of a parameter.
+        Returns None if the parameter is not defined.
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    def set_param(self, name: str, value: Any) -> None:
+        """Set the value of a parameter.
+        Adds the parameter if it is not already defined.
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    def copy_component(self, *args, **kwargs) -> "GraphComponent":
+        """Get a new instance of this graph component type with the same name, id and parameters."""
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def neighbors(self) -> Iterable["GraphComponent"]:
+        """Get all components that are directly connected to this operation."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def traverse(self) -> Generator["GraphComponent", None, None]:
+        """Get a generator that recursively iterates through all components that are connected to this operation,
+        as well as the ones that they are connected to.
+        """
         raise NotImplementedError
 
 
 class AbstractGraphComponent(GraphComponent):
     """Abstract Graph Component class which is a component of a signal flow graph.
-
     TODO: More info.
     """
 
     _name: Name
+    _graph_id: GraphID
+    _parameters: Dict[str, Any]
 
     def __init__(self, name: Name = ""):
         self._name = name
+        self._graph_id = ""
+        self._parameters = {}
+
+    def __str__(self):
+        return f"id: {self.graph_id if self.graph_id else 'no_id'}, \tname: {self.name if self.name else 'no_name'}" + \
+            "".join((f", \t{key}: {str(param)}" for key, param in self._parameters.items()))
 
     @property
     def name(self) -> Name:
@@ -52,3 +116,41 @@ class AbstractGraphComponent(GraphComponent):
     @name.setter
     def name(self, name: Name) -> None:
         self._name = name
+
+    @property
+    def graph_id(self) -> GraphID:
+        return self._graph_id
+
+    @graph_id.setter
+    def graph_id(self, graph_id: GraphID) -> None:
+        self._graph_id = graph_id
+
+    @property
+    def params(self) -> Mapping[str, Any]:
+        return self._parameters.copy()
+
+    def param(self, name: str) -> Any:
+        return self._parameters.get(name)
+
+    def set_param(self, name: str, value: Any) -> None:
+        self._parameters[name] = value
+
+    def copy_component(self, *args, **kwargs) -> GraphComponent:
+        new_component = self.__class__(*args, **kwargs)
+        new_component.name = copy(self.name)
+        new_component.graph_id = copy(self.graph_id)
+        for name, value in self.params.items():
+            new_component.set_param(copy(name), deepcopy(value))  # pylint: disable=no-member
+        return new_component
+
+    def traverse(self) -> Generator[GraphComponent, None, None]:
+        # Breadth first search.
+        visited = {self}
+        fontier = deque([self])
+        while fontier:
+            component = fontier.popleft()
+            yield component
+            for neighbor in component.neighbors:
+                if neighbor not in visited:
+                    visited.add(neighbor)
+                    fontier.append(neighbor)

b_asic/graph_id.py

-"""@package docstring
-B-ASIC Graph ID module for handling IDs of different objects in a graph.
-TODO: More info
-"""
-
-from collections import defaultdict
-from typing import NewType, DefaultDict
-
-GraphID = NewType("GraphID", str)
-GraphIDType = NewType("GraphIDType", str)
-GraphIDNumber = NewType("GraphIDNumber", int)
-
-
-class GraphIDGenerator:
-    """A class that generates Graph IDs for objects."""
-
-    _next_id_number: DefaultDict[GraphIDType, GraphIDNumber]
-
-    def __init__(self):
-        self._next_id_number = defaultdict(lambda: 1)       # Initalises every key element to 1
-
-    def get_next_id(self, graph_id_type: GraphIDType) -> GraphID:
-        """Return the next graph id for a certain graph id type."""
-        graph_id = graph_id_type + str(self._next_id_number[graph_id_type])
-        self._next_id_number[graph_id_type] += 1            # Increase the current id number
-        return graph_id

b_asic/port.py

@@ -4,12 +4,15 @@ TODO: More info.
 """
 
 from abc import ABC, abstractmethod
-from typing import NewType, Optional, List
+from copy import copy
+from typing import Optional, List, Iterable, TYPE_CHECKING
 
-from b_asic.operation import Operation
 from b_asic.signal import Signal
+from b_asic.graph_component import Name
+
+if TYPE_CHECKING:
+    from b_asic.operation import Operation
 
-PortIndex = NewType("PortIndex", int)
 
 class Port(ABC):
     """Port Interface.
@@ -19,59 +22,45 @@ class Port(ABC):
 
     @property
     @abstractmethod
-    def operation(self) -> Operation:
+    def operation(self) -> "Operation":
         """Return the connected operation."""
         raise NotImplementedError
 
     @property
     @abstractmethod
-    def index(self) -> PortIndex:
-        """Return the unique PortIndex."""
+    def index(self) -> int:
+        """Return the index of the port."""
         raise NotImplementedError
 
     @property
     @abstractmethod
-    def signals(self) -> List[Signal]:
-        """Return a list of all connected signals."""
+    def latency_offset(self) -> int:
+        """Get the latency_offset of the port."""
         raise NotImplementedError
 
+    @latency_offset.setter
     @abstractmethod
-    def signal(self, i: int = 0) -> Signal:
-        """Return the connected signal at index i.
-
-        Keyword argumens:
-        i: integer index of the signal requsted.
-        """
+    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 connected_ports(self) -> List["Port"]:
-        """Return a list of all connected Ports."""
-        raise NotImplementedError
-
     @abstractmethod
     def signal_count(self) -> int:
         """Return the number of connected signals."""
         raise NotImplementedError
 
+    @property
     @abstractmethod
-    def connect(self, port: "Port") -> Signal:
-        """Create and return a signal that is connected to this port and the entered
-        port and connect this port to the signal and the entered port to the signal."""
+    def signals(self) -> Iterable[Signal]:
+        """Return all connected signals."""
         raise NotImplementedError
 
     @abstractmethod
     def add_signal(self, signal: Signal) -> None:
         """Connect this port to the entered signal. If the entered signal isn't connected to
-        this port then connect the entered signal to the port aswell."""
-        raise NotImplementedError
-
-    @abstractmethod
-    def disconnect(self, port: "Port") -> None:
-        """Disconnect the entered port from the port by removing it from the ports signal.
-        If the entered port is still connected to this ports signal then disconnect the entered
-        port from the signal aswell."""
+        this port then connect the entered signal to the port aswell.
+        """
         raise NotImplementedError
 
     @abstractmethod
@@ -97,21 +86,43 @@ class AbstractPort(Port):
     Handles functionality for port id and saves the connection to the parent operation.
     """
 
+    _operation: "Operation"
     _index: int
-    _operation: Operation
+    _latency_offset: Optional[int]
 
-    def __init__(self, index: int, operation: Operation):
-        self._index = index
+    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:
+    def operation(self) -> "Operation":
         return self._operation
 
     @property
-    def index(self) -> PortIndex:
+    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.
+    TODO: More info.
+    """
+
+    @property
+    @abstractmethod
+    def source(self) -> "OutputPort":
+        """Get the main source port provided by this object."""
+        raise NotImplementedError
+
 
 class InputPort(AbstractPort):
     """Input port.
@@ -120,104 +131,82 @@ class InputPort(AbstractPort):
 
     _source_signal: Optional[Signal]
 
-    def __init__(self, port_id: PortIndex, operation: Operation):
-        super().__init__(port_id, operation)
+    def __init__(self, operation: "Operation", index: int):
+        super().__init__(operation, index)
         self._source_signal = None
 
     @property
-    def signals(self) -> List[Signal]:
-        return [] if self._source_signal is None else [self._source_signal]
-
-    def signal(self, i: int = 0) -> Signal:
-        assert 0 <= i < self.signal_count(), "Signal index out of bound."
-        assert self._source_signal is not None, "No Signal connect to InputPort."
-        return self._source_signal
-
-    @property
-    def connected_ports(self) -> List[Port]:
-        return [] if self._source_signal is None or self._source_signal.source is None \
-            else [self._source_signal.source]
-
     def signal_count(self) -> int:
         return 0 if self._source_signal is None else 1
 
-    def connect(self, port: "OutputPort") -> Signal:
-        assert self._source_signal is None, "Connecting new port to already connected input port."
-        return Signal(port, self)        # self._source_signal is set by the signal constructor
+    @property
+    def signals(self) -> Iterable[Signal]:
+        return [] if self._source_signal is None else [self._source_signal]
 
     def add_signal(self, signal: Signal) -> None:
-        assert self._source_signal is None, "Connecting new port to already connected input port."
-        self._source_signal: Signal = signal
-        if self is not signal.destination:
-            # Connect this inputport as destination for this signal if it isn't already.
-            signal.set_destination(self)
-
-    def disconnect(self, port: "OutputPort") -> None:
-        assert self._source_signal.source is port, "The entered port is not connected to this port."
-        self._source_signal.remove_source()
+        assert self._source_signal is None, "Input port may have only one signal added."
+        assert signal is not self._source_signal, "Attempted to add already connected signal."
+        self._source_signal = signal
+        signal.set_destination(self)
 
     def remove_signal(self, signal: Signal) -> None:
-        old_signal: Signal = self._source_signal
+        assert signal is self._source_signal, "Attempted to remove signal that is not connected."
         self._source_signal = None
-        if self is old_signal.destination:
-            # Disconnect the dest of the signal if this inputport currently is the dest
-            old_signal.remove_destination()
+        signal.remove_destination()
 
     def clear(self) -> None:
-        self.remove_signal(self._source_signal)
+        if self._source_signal is not None:
+            self.remove_signal(self._source_signal)
+
+    @property
+    def connected_source(self) -> Optional["OutputPort"]:
+        """Get the output port that is currently connected to this input port,
+        or None if it is unconnected.
+        """
+        return None if self._source_signal is None else self._source_signal.source
 
-class OutputPort(AbstractPort):
+    def connect(self, src: SignalSourceProvider, name: Name = "") -> Signal:
+        """Connect the provided signal source to this input port by creating a new signal.
+        Returns the new signal.
+        """
+        assert self._source_signal is None, "Attempted to connect already connected input port."
+        # self._source_signal is set by the signal constructor.
+        return Signal(source=src.source, destination=self, name=name)
+
+
+class OutputPort(AbstractPort, SignalSourceProvider):
     """Output port.
     TODO: More info.
     """
 
     _destination_signals: List[Signal]
 
-    def __init__(self, port_id: PortIndex, operation: Operation):
-        super().__init__(port_id, operation)
+    def __init__(self, operation: "Operation", index: int):
+        super().__init__(operation, index)
         self._destination_signals = []
 
     @property
-    def signals(self) -> List[Signal]:
-        return self._destination_signals.copy()
-
-    def signal(self, i: int = 0) -> Signal:
-        assert 0 <= i < self.signal_count(), "Signal index out of bounds."
-        return self._destination_signals[i]
-
-    @property
-    def connected_ports(self) -> List[Port]:
-        return [signal.destination for signal in self._destination_signals \
-            if signal.destination is not None]
-
     def signal_count(self) -> int:
         return len(self._destination_signals)
 
-    def connect(self, port: InputPort) -> Signal:
-        return Signal(self, port)      # Signal is added to self._destination_signals in signal constructor
+    @property
+    def signals(self) -> Iterable[Signal]:
+        return self._destination_signals
 
     def add_signal(self, signal: Signal) -> None:
-        assert signal not in self.signals, \
-                "Attempting to connect to Signal already connected."
+        assert signal not in self._destination_signals, "Attempted to add already connected signal."
         self._destination_signals.append(signal)
-        if self is not signal.source:
-            # Connect this outputport to the signal if it isn't already
-            signal.set_source(self)
-
-    def disconnect(self, port: InputPort) -> None:
-        assert port in self.connected_ports, "Attempting to disconnect port that isn't connected."
-        for sig in self._destination_signals:
-            if sig.destination is port:
-                sig.remove_destination()
-                break
+        signal.set_source(self)
 
     def remove_signal(self, signal: Signal) -> None:
-        i: int = self._destination_signals.index(signal)
-        old_signal: Signal = self._destination_signals[i]
-        del self._destination_signals[i]
-        if self is old_signal.source:
-            old_signal.remove_source()
+        assert signal in self._destination_signals, "Attempted to remove signal that is not connected."
+        self._destination_signals.remove(signal)
+        signal.remove_source()
 
     def clear(self) -> None:
-        for signal in self._destination_signals:
+        for signal in copy(self._destination_signals):
             self.remove_signal(signal)
+
+    @property
+    def source(self) -> "OutputPort":
+        return self

b_asic/precedence_chart.py

-"""@package docstring
-B-ASIC Precedence Chart Module.
-TODO: More info.
-"""
-
-from b_asic.signal_flow_graph import SFG
-
-
-class PrecedenceChart:
-    """Precedence chart constructed from a signal flow graph.
-    TODO: More info.
-    """
-
-    sfg: SFG
-    # TODO: More members.
-
-    def __init__(self, sfg: SFG):
-        self.sfg = sfg
-        # TODO: Implement.
-
-    # TODO: More stuff.

b_asic/schema.py

 """@package docstring
-B-ASIC Schema Module.
-TODO: More info.
+This module contains the Schema class.
+TODO: More info
 """
 
-from b_asic.precedence_chart import PrecedenceChart
+from io import StringIO
+from typing import Dict, List, Set
+import math
+
+from b_asic.signal_flow_graph import SFG
+from b_asic.graph_component import GraphID
+from b_asic.operation import Operation
 
 
 class Schema:
-    """Schema constructed from a precedence chart.
-    TODO: More info.
-    """
+    """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
+    _max_end_op_id: GraphID
+    _max_end_time: int
+    _non_schedulable_ops: Set[GraphID]
+
+    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 ValueError(f"No algorithm with name: {scheduling_alg} defined.")
+
+        # Set latest time of the schema.
+        self._latest_op_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:
+                if op_start_time + outport.latency_offset > self._latest_op_time:
+                    self._max_end_op_id = op_id
+                    self._max_end_time = op_start_time + outport.latency_offset
+
+        if not self._cyclic:
+            if schedule_time is None:
+                self.set_schedule_time(self._max_end_time)
+            else:
+                self.set_schedule_time(schedule_time)
+        else:
+            raise NotImplementedError("Cyclic schedules not implemented yet.")
+
+    def __str__(self):
+        string_io = StringIO()
+        for op in self._sfg.get_operations_topological_order():
+            if op.graph_id in self._start_times:
+                string_io.write("name: " + op.name + ", \tid: " + op.graph_id + ", \t")
+                string_io.write("start_time: " + str(self._start_times[op.graph_id]) + ", \t")
+                string_io.write("backwards_slack: " + str(self.backwards_slack(op.graph_id)) + ", \t")
+                string_io.write("forwards_slack: " + str(self.forwards_slack(op.graph_id)) + "\n")
+        return string_io.getvalue()
+
+    @property
+    def schedule_time(self) -> int:
+        """Get the schedule time of the Schema."""
+        return self._schedule_time
+
+    def set_schedule_time(self, schedule_time: int) -> None:
+        """Set the schedule time to the specified value, if the specified value is shorter then 
+        what is required by the schedule then raises a ValueError."""
+        if not self._cyclic:
+            if schedule_time >= self._max_end_time:
+                self._schedule_time = schedule_time
+            else:
+                raise ValueError("Specified schedule time is shorter then the max length of the ")
+        else:
+            raise NotImplementedError("Cyclic schedules not implemented yet.")
+
+    def get_start_time_of_operation(self, op_id: GraphID) -> int:
+        """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 get_laps_of_operation_input_port(self, op_id: GraphID, input_index: int) -> int:
+        """Get the laps of input port with the entered index of the operation with the specified op_id."""
+        assert op_id in self._laps, "No operation with the specidied op_id in this schema."
+        assert input_index >= 0 and input_index < len(
+            self._laps[op_id]), "Input index for the specified operation out of bounds."
+        return self._laps[op_id][input_index]
+
+    def change_start_time_of_operation(self, op_id: GraphID, relative_change: int = None, new_start_time: int = None) -> None:
+        """Changes the start time of the operation with the specified op_id either by the 
+        relative_change if specified or to the new_start_time if that is specified. If both 
+        are specified then the new_start_time value is used.
+        A negative relative_change represents decreasing the operations start time.
+        If the change to the start time would break dependencies then the change is cancelled
+        and a message is printed to the user in the terminal.
+        """
+        assert op_id in self._start_times, "No operation with the specified op_id in this schema."
+        if new_start_time is not None:
+            relative_change = new_start_time - self._start_times[op_id]
+
+        if relative_change is not None:
+            if relative_change > 0:
+                forwards_slack = self.forwards_slack(op_id)
+                if relative_change > forwards_slack:
+                    print(f"The specified relative change of {relative_change} is larger than the forwards \
+                        slack of the operation of {forwards_slack}.")
+                    return
+            else:
+                backwards_slack = self.backwards_slack(op_id)
+                if abs(relative_change) > backwards_slack:
+                    print(f"The specified relative change of {abs(relative_change)} is larger than the forwards \
+                        slack of the operation of {forwards_slack}.")
+                    return
+
+            self._start_times[op_id] += relative_change
+
+        else:
+            raise ValueError("At least one of the parameters relative_change or new_start_time has to be specified.")
+
+    def backwards_slack(self, op_id: GraphID) -> int:
+        return 0
+        op_start_time = self.get_start_time_of_operation(op_id)
+        min_backwards_slack = math.inf
+        for i, inp in enumerate(self._sfg.find_by_id(op_id).inputs):
+            inp_time = op_start_time + inp.latency_offset
+
+            source_op_time = self.get_start_time_of_operation(inp.connected_source.operation.graph_id)
+            source_outp_time = source_op_time + inp.connected_source.latency_offset
+
+    def forwards_slack(self, op_id: GraphID) -> int:
+        return 0
+        assert op_id in self._start_times, "No operation with specified op_id in this schema."
+
+    def _schedule_asap(self) -> None:
+        pl = self._sfg.get_precedence_list()
+
+        if len(pl) < 2:
+            print("Empty signal flow graph cannot be scheduled.")
+            return
+
+        self._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:
+                        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 self._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."
 
-    pc: PrecedenceChart
-    # TODO: More members.
+                            source_end_time = source_op_time + source_port.latency_offset
 
-    def __init__(self, pc: PrecedenceChart):
-        self.pc = pc
-        # TODO: Implement.
+                        op_start_time_from_in = source_end_time - inport.latency_offset
+                        op_start_time = max(op_start_time, op_start_time_from_in)
 
-    # TODO: More stuff.
+                    self._start_times[op.graph_id] = op_start_time

b_asic/signal.py

 """@package docstring
 B-ASIC Signal Module.
 """
-from typing import Optional, TYPE_CHECKING
+from typing import Optional, Iterable, TYPE_CHECKING
 
-from b_asic.graph_component import AbstractGraphComponent, TypeName, Name
+from b_asic.graph_component import GraphComponent, AbstractGraphComponent, TypeName, Name
 
 if TYPE_CHECKING:
     from b_asic.port import InputPort, OutputPort
@@ -12,30 +12,34 @@ if TYPE_CHECKING:
 class Signal(AbstractGraphComponent):
     """A connection between two ports."""
 
-    _source: "OutputPort"
-    _destination: "InputPort"
-
-    def __init__(self, source: Optional["OutputPort"] = None, \
-                destination: Optional["InputPort"] = None, name: Name = ""):
+    _source: Optional["OutputPort"]
+    _destination: Optional["InputPort"]
 
+    def __init__(self, source: Optional["OutputPort"] = None, destination: Optional["InputPort"] = None, bits: Optional[int] = None, name: Name = ""):
         super().__init__(name)
-
-        self._source = source
-        self._destination = destination
-
+        self._source = None
+        self._destination = None
         if source is not None:
             self.set_source(source)
-
         if destination is not None:
             self.set_destination(destination)
+        self.set_param("bits", bits)
+
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "s"
 
     @property
-    def source(self) -> "OutputPort":
+    def neighbors(self) -> Iterable[GraphComponent]:
+        return [p.operation for p in [self.source, self.destination] if p is not None]
+
+    @property
+    def source(self) -> Optional["OutputPort"]:
         """Return the source OutputPort of the signal."""
         return self._source
 
     @property
-    def destination(self) -> "InputPort":
+    def destination(self) -> Optional["InputPort"]:
         """Return the destination "InputPort" of the signal."""
         return self._destination
 
@@ -47,11 +51,11 @@ class Signal(AbstractGraphComponent):
         Keyword arguments:
         - src: OutputPort to connect as source to the signal.
         """
-        self.remove_source()
-        self._source = src
-        if self not in src.signals:
-            # If the new source isn't connected to this signal then connect it.
-            src.add_signal(self)
+        if src is not self._source:
+            self.remove_source()
+            self._source = src
+            if self not in src.signals:
+                src.add_signal(self)
 
     def set_destination(self, dest: "InputPort") -> None:
         """Disconnect the previous destination InputPort of the signal and
@@ -61,36 +65,44 @@ class Signal(AbstractGraphComponent):
         Keywords argments:
         - dest: InputPort to connect as destination to the signal.
         """
-        self.remove_destination()
-        self._destination = dest
-        if self not in dest.signals:
-            # If the new destination isn't connected to tis signal then connect it.
-            dest.add_signal(self)
-
-    @property
-    def type_name(self) -> TypeName:
-        return "s"
+        if dest is not self._destination:
+            self.remove_destination()
+            self._destination = dest
+            if self not in dest.signals:
+                dest.add_signal(self)
 
     def remove_source(self) -> None:
         """Disconnect the source OutputPort of the signal. If the source port
         still is connected to this signal then also disconnect the source port."""
-        if self._source is not None:
-            old_source: "OutputPort" = self._source
+        src = self._source
+        if src is not None:
             self._source = None
-            if self in old_source.signals:
-                # If the old destination port still is connected to this signal, then disconnect it.
-                old_source.remove_signal(self)
+            if self in src.signals:
+                src.remove_signal(self)
 
     def remove_destination(self) -> None:
         """Disconnect the destination InputPort of the signal."""
-        if self._destination is not None:
-            old_destination: "InputPort" = self._destination
+        dest = self._destination
+        if dest is not None:
             self._destination = None
-            if self in old_destination.signals:
-                # If the old destination port still is connected to this signal, then disconnect it.
-                old_destination.remove_signal(self)
+            if self in dest.signals:
+                dest.remove_signal(self)
 
-    def is_connected(self) -> bool:
-        """Returns true if the signal is connected to both a source and a destination,
+    def dangling(self) -> bool:
+        """Returns true if the signal is missing either a source or a destination,
         else false."""
-        return self._source is not None and self._destination is not None
+        return self._source is None or self._destination is None
+
+    @property
+    def bits(self) -> Optional[int]:
+        """Get the number of bits that this operations using this signal as an input should truncate received values to.
+        None = unlimited."""
+        return self.param("bits")
+
+    @bits.setter
+    def bits(self, bits: Optional[int]) -> None:
+        """Set the number of bits that operations using this signal as an input should truncate received values to.
+        None = unlimited."""
+        assert bits is None or (isinstance(bits, int)
+                                and bits >= 0), "Bits must be non-negative."
+        self.set_param("bits", bits)

b_asic/simulation.py

@@ -3,33 +3,116 @@ B-ASIC Simulation Module.
 TODO: More info.
 """
 
+from collections import defaultdict
 from numbers import Number
-from typing import List
+from typing import List, Dict, DefaultDict, Callable, Sequence, Mapping, Union, Optional
 
+from b_asic.operation import OutputKey, OutputMap
+from b_asic.signal_flow_graph import SFG
 
-class OperationState:
-    """Simulation state of an operation.
+
+InputProvider = Union[Number, Sequence[Number], Callable[[int], Number]]
+
+
+class Simulation:
+    """Simulation.
     TODO: More info.
     """
 
-    output_values: List[Number]
-    iteration: int
+    _sfg: SFG
+    _results: DefaultDict[int, Dict[str, Number]]
+    _registers: Dict[str, Number]
+    _iteration: int
+    _input_functions: Sequence[Callable[[int], Number]]
+    _current_input_values: Sequence[Number]
+    _latest_output_values: Sequence[Number]
+    _save_results: bool
 
-    def __init__(self):
-        self.output_values = []
-        self.iteration = 0
+    def __init__(self, sfg: SFG, input_providers: Optional[Sequence[Optional[InputProvider]]] = None, save_results: bool = False):
+        self._sfg = sfg
+        self._results = defaultdict(dict)
+        self._registers = {}
+        self._iteration = 0
+        self._input_functions = [
+            lambda _: 0 for _ in range(self._sfg.input_count)]
+        self._current_input_values = [0 for _ in range(self._sfg.input_count)]
+        self._latest_output_values = [0 for _ in range(self._sfg.output_count)]
+        self._save_results = save_results
+        if input_providers is not None:
+            self.set_inputs(input_providers)
 
+    def set_input(self, index: int, input_provider: InputProvider) -> None:
+        """Set the input function used to get values for the specific input at the given index to the internal SFG."""
+        if index < 0 or index >= len(self._input_functions):
+            raise IndexError(
+                f"Input index out of range (expected 0-{len(self._input_functions) - 1}, got {index})")
+        if callable(input_provider):
+            self._input_functions[index] = input_provider
+        elif isinstance(input_provider, Number):
+            self._input_functions[index] = lambda _: input_provider
+        else:
+            self._input_functions[index] = lambda n: input_provider[n]
 
-class SimulationState:
-    """Simulation state.
-    TODO: More info.
-    """
+    def set_inputs(self, input_providers: Sequence[Optional[InputProvider]]) -> None:
+        """Set the input functions used to get values for the inputs to the internal SFG."""
+        if len(input_providers) != self._sfg.input_count:
+            raise ValueError(
+                f"Wrong number of inputs supplied to simulation (expected {self._sfg.input_count}, got {len(input_providers)})")
+        self._input_functions = [None for _ in range(self._sfg.input_count)]
+        for index, input_provider in enumerate(input_providers):
+            if input_provider is not None:
+                self.set_input(index, input_provider)
+
+    @property
+    def save_results(self) -> bool:
+        """Get the flag that determines if the results of ."""
+        return self._save_results
+
+    @save_results.setter
+    def save_results(self, save_results) -> None:
+        self._save_results = save_results
+
+    def run(self) -> Sequence[Number]:
+        """Run one iteration of the simulation and return the resulting output values."""
+        return self.run_for(1)
+
+    def run_until(self, iteration: int) -> Sequence[Number]:
+        """Run the simulation until its iteration is greater than or equal to the given iteration
+        and return the resulting output values.
+        """
+        while self._iteration < iteration:
+            self._current_input_values = [self._input_functions[i](
+                self._iteration) for i in range(self._sfg.input_count)]
+            self._latest_output_values = self._sfg.evaluate_outputs(
+                self._current_input_values, self._results[self._iteration], self._registers)
+            if not self._save_results:
+                del self._results[self.iteration]
+            self._iteration += 1
+        return self._latest_output_values
+
+    def run_for(self, iterations: int) -> Sequence[Number]:
+        """Run a given number of iterations of the simulation and return the resulting output values."""
+        return self.run_until(self._iteration + iterations)
+
+    @property
+    def iteration(self) -> int:
+        """Get the current iteration number of the simulation."""
+        return self._iteration
 
-    # operation_states: Dict[OperationId, OperationState]
-    iteration: int
+    @property
+    def results(self) -> Mapping[int, OutputMap]:
+        """Get a mapping of all results, including intermediate values, calculated for each iteration up until now.
+        The outer mapping maps from iteration number to value mapping. The value mapping maps output port identifiers to values.
+        Example: {0: {"c1": 3, "c2": 4, "bfly1.0": 7, "bfly1.1": -1, "0": 7}}
+        """
+        return self._results
 
-    def __init__(self):
-        self.operation_states = {}
-        self.iteration = 0
+    def clear_results(self) -> None:
+        """Clear all results that were saved until now."""
+        self._results.clear()
 
-    # TODO: More stuff.
+    def clear_state(self) -> None:
+        """Clear all current state of the simulation, except for the results and iteration."""
+        self._registers.clear()
+        self._current_input_values = [0 for _ in range(self._sfg.input_count)]
+        self._latest_output_values = [0 for _ in range(self._sfg.output_count)]

b_asic/special_operations.py

+"""@package docstring
+B-ASIC Special Operations Module.
+TODO: More info.
+"""
+
+from numbers import Number
+from typing import Optional, Sequence
+
+from b_asic.operation import AbstractOperation, RegisterMap, MutableOutputMap, MutableRegisterMap
+from b_asic.graph_component import Name, TypeName
+from b_asic.port import SignalSourceProvider
+
+
+class Input(AbstractOperation):
+    """Input operation.
+    TODO: More info.
+    """
+
+    def __init__(self, name: Name = ""):
+        super().__init__(input_count=0, output_count=1, name=name)
+        self.set_param("value", 0)
+
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "in"
+
+    def evaluate(self):
+        return self.param("value")
+
+    @property
+    def value(self) -> Number:
+        """Get the current value of this input."""
+        return self.param("value")
+
+    @value.setter
+    def value(self, value: Number) -> None:
+        """Set the current value of this input."""
+        self.set_param("value", value)
+
+
+class Output(AbstractOperation):
+    """Output operation.
+    TODO: More info.
+    """
+
+    def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
+        super().__init__(input_count=1, output_count=0,
+                         name=name, input_sources=[src0])
+
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "out"
+
+    def evaluate(self, _):
+        return None
+
+
+class Register(AbstractOperation):
+    """Unit delay operation.
+    TODO: More info.
+    """
+
+    def __init__(self, src0: Optional[SignalSourceProvider] = None, initial_value: Number = 0, name: Name = ""):
+        super().__init__(input_count=1, output_count=1,
+                         name=name, input_sources=[src0])
+        self.set_param("initial_value", initial_value)
+
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "reg"
+
+    def evaluate(self, a):
+        return self.param("initial_value")
+
+    def current_output(self, index: int, registers: Optional[RegisterMap] = None, prefix: str = "") -> Optional[Number]:
+        if registers is not None:
+            return registers.get(self.key(index, prefix), self.param("initial_value"))
+        return self.param("initial_value")
+
+    def evaluate_output(self, index: int, input_values: Sequence[Number], results: Optional[MutableOutputMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = "") -> Number:
+        if index != 0:
+            raise IndexError(
+                f"Output index out of range (expected 0-0, got {index})")
+        if len(input_values) != 1:
+            raise ValueError(
+                f"Wrong number of inputs supplied to SFG for evaluation (expected 1, got {len(input_values)})")
+
+        key = self.key(index, prefix)
+        value = self.param("initial_value")
+        if registers is not None:
+            value = registers.get(key, value)
+            registers[key] = self.truncate_inputs(input_values)[0]
+        if results is not None:
+            results[key] = value
+        return value

setup.py

@@ -36,9 +36,9 @@ class CMakeBuild(build_ext):
 
         if not os.path.exists(self.build_temp):
             os.makedirs(self.build_temp)
-        
+
         env = os.environ.copy()
-        
+
         print(f"=== Configuring {ext.name} ===")
         print(f"Temp dir: {self.build_temp}")
         print(f"Output dir: {cmake_output_dir}")
@@ -71,10 +71,12 @@ setuptools.setup(
     install_requires = [
         "pybind11>=2.3.0",
         "numpy",
-        "install_qt_binding"
+        "pyside2",
+        "graphviz",
+        "matplotlib"
     ],
     packages = ["b_asic"],
     ext_modules = [CMakeExtension("b_asic")],
     cmdclass = {"build_ext": CMakeBuild},
     zip_safe = False
-)
\ No newline at end of file
+)