Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • da/B-ASIC
  • lukja239/B-ASIC
  • robal695/B-ASIC
3 results
Show changes
Commits on Source (95)
Expand all Hide whitespace changes
Inline Side-by-side
Showing
with 1658 additions and 740 deletions
.gitignore
View file @ 58d9907d
.vs/
.vscode/
build*/
bin*/
logs/
dist/
CMakeLists.txt.user*
*.autosave
*.creator
*.creator.user*
\#*\#
/.emacs.desktop
/.emacs.desktop.lock
*.elc
auto-save-list
tramp
.\#*
*~
.fuse_hudden*
.directory
.Trash-*
.nfs*
Thumbs.db
Thumbs.db:encryptable
ehthumbs.db
ehthumbs_vista.db
$RECYCLE.BIN/
*.stackdump
[Dd]esktop.ini
*.egg-info
__pycache__/
env/
.vs/
.vscode/
build*/
bin*/
logs/
dist/
CMakeLists.txt.user*
*.autosave
*.creator
*.creator.user*
\#*\#
/.emacs.desktop
/.emacs.desktop.lock
*.elc
auto-save-list
tramp
.\#*
*~
.fuse_hudden*
.directory
.Trash-*
.nfs*
Thumbs.db
Thumbs.db:encryptable
ehthumbs.db
ehthumbs_vista.db
$RECYCLE.BIN/
*.stackdump
[Dd]esktop.ini
*.egg-info
__pycache__/
env/
venv/
\ No newline at end of file
GUI/main_window.py 0 → 100644
View file @ 58d9907d
"""@package docstring
B-ASIC GUI Module.
This python file is an example of how a GUI can be implemented
using buttons and textboxes.
"""
import sys
from PyQt5.QtWidgets import QApplication, QWidget, QMainWindow, QLabel, QAction,\
QStatusBar, QMenuBar, QLineEdit, QPushButton
from PyQt5.QtCore import Qt
from PyQt5.QtGui import QIcon, QFont, QPainter, QPen
class DragButton(QPushButton):
"""How to create a dragbutton"""
def mousePressEvent(self, event):
self._mouse_press_pos = None
self._mouse_move_pos = None
if event.button() == Qt.LeftButton:
self._mouse_press_pos = event.globalPos()
self._mouse_move_pos = event.globalPos()
super(DragButton, self).mousePressEvent(event)
def mouseMoveEvent(self, event):
if event.buttons() == Qt.LeftButton:
cur_pos = self.mapToGlobal(self.pos())
global_pos = event.globalPos()
diff = global_pos - self._mouse_move_pos
new_pos = self.mapFromGlobal(cur_pos + diff)
self.move(new_pos)
self._mouse_move_pos = global_pos
super(DragButton, self).mouseMoveEvent(event)
def mouseReleaseEvent(self, event):
if self._mouse_press_pos is not None:
moved = event.globalPos() - self._mouse_press_pos
if moved.manhattanLength() > 3:
event.ignore()
return
super(DragButton, self).mouseReleaseEvent(event)
class SubWindow(QWidget):
"""Creates a sub window """
def create_window(self, window_width, window_height):
"""Creates a window
"""
parent = None
super(SubWindow, self).__init__(parent)
self.setWindowFlags(Qt.WindowStaysOnTopHint)
self.resize(window_width, window_height)
class MainWindow(QMainWindow):
"""Main window for the program"""
def __init__(self, *args, **kwargs):
super(MainWindow, self).__init__(*args, **kwargs)
self.setWindowTitle(" ")
self.setWindowIcon(QIcon('small_logo.png'))
# Menu buttons
test_button = QAction("Test", self)
exit_button = QAction("Exit", self)
exit_button.setShortcut("Ctrl+Q")
exit_button.triggered.connect(self.exit_app)
edit_button = QAction("Edit", self)
edit_button.setStatusTip("Open edit menu")
edit_button.triggered.connect(self.on_edit_button_click)
view_button = QAction("View", self)
view_button.setStatusTip("Open view menu")
view_button.triggered.connect(self.on_view_button_click)
menu_bar = QMenuBar()
menu_bar.setStyleSheet("background-color:rgb(222, 222, 222)")
self.setMenuBar(menu_bar)
file_menu = menu_bar.addMenu("&File")
file_menu.addAction(exit_button)
file_menu.addSeparator()
file_menu.addAction(test_button)
edit_menu = menu_bar.addMenu("&Edit")
edit_menu.addAction(edit_button)
edit_menu.addSeparator()
view_menu = menu_bar.addMenu("&View")
view_menu.addAction(view_button)
self.setStatusBar(QStatusBar(self))
def on_file_button_click(self):
print("File")
def on_edit_button_click(self):
print("Edit")
def on_view_button_click(self):
print("View")
def exit_app(self, checked):
QApplication.quit()
def clicked(self):
print("Drag button clicked")
def add_drag_buttons(self):
"""Adds draggable buttons"""
addition_button = DragButton("Addition", self)
addition_button.move(10, 130)
addition_button.setFixedSize(70, 20)
addition_button.clicked.connect(self.create_sub_window)
addition_button2 = DragButton("Addition", self)
addition_button2.move(10, 130)
addition_button2.setFixedSize(70, 20)
addition_button2.clicked.connect(self.create_sub_window)
subtraction_button = DragButton("Subtraction", self)
subtraction_button.move(10, 170)
subtraction_button.setFixedSize(70, 20)
subtraction_button.clicked.connect(self.create_sub_window)
subtraction_button2 = DragButton("Subtraction", self)
subtraction_button2.move(10, 170)
subtraction_button2.setFixedSize(70, 20)
subtraction_button2.clicked.connect(self.create_sub_window)
multiplication_button = DragButton("Multiplication", self)
multiplication_button.move(10, 210)
multiplication_button.setFixedSize(70, 20)
multiplication_button.clicked.connect(self.create_sub_window)
multiplication_button2 = DragButton("Multiplication", self)
multiplication_button2.move(10, 210)
multiplication_button2.setFixedSize(70, 20)
multiplication_button2.clicked.connect(self.create_sub_window)
def paintEvent(self, e):
# Temporary black box for operations
painter = QPainter(self)
painter.setPen(QPen(Qt.black, 5, Qt.SolidLine))
painter.drawRect(0, 110, 100, 400)
# Temporary arrow resembling a signal
painter.setRenderHint(QPainter.Antialiasing)
painter.setPen(Qt.black)
painter.setBrush(Qt.white)
painter.drawLine(300, 200, 400, 200)
painter.drawLine(400, 200, 395, 195)
painter.drawLine(400, 200, 395, 205)
def create_sub_window(self):
""" Example of how to create a sub window
"""
self.sub_window = SubWindow()
self.sub_window.create_window(400, 300)
self.sub_window.setWindowTitle("Properties")
self.sub_window.properties_label = QLabel(self.sub_window)
self.sub_window.properties_label.setText('Properties')
self.sub_window.properties_label.setFixedWidth(400)
self.sub_window.properties_label.setFont(QFont('SansSerif', 14, QFont.Bold))
self.sub_window.properties_label.setAlignment(Qt.AlignCenter)
self.sub_window.name_label = QLabel(self.sub_window)
self.sub_window.name_label.setText('Name:')
self.sub_window.name_label.move(20, 40)
self.sub_window.name_line = QLineEdit(self.sub_window)
self.sub_window.name_line.setPlaceholderText("Write a name here")
self.sub_window.name_line.move(70, 40)
self.sub_window.name_line.resize(100, 20)
self.sub_window.id_label = QLabel(self.sub_window)
self.sub_window.id_label.setText('Id:')
self.sub_window.id_label.move(20, 70)
self.sub_window.id_line = QLineEdit(self.sub_window)
self.sub_window.id_line.setPlaceholderText("Write an id here")
self.sub_window.id_line.move(70, 70)
self.sub_window.id_line.resize(100, 20)
self.sub_window.show()
if __name__ == "__main__":
app = QApplication(sys.argv)
window = MainWindow()
window.add_drag_buttons()
window.resize(960, 720)
window.show()
app.exec_()
b_asic/__init__.py
View file @ 58d9907d
......@@ -4,7 +4,6 @@ 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 *
......@@ -12,3 +11,4 @@ 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 *
b_asic/core_operations.py
View file @ 58d9907d
......@@ -4,43 +4,39 @@ TODO: More info.
"""
from numbers import Number
from typing import Any
from typing import Optional
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
@property
def type_name(self) -> 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,81 @@ 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)]
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
def evaluate(self, a, b):
return a + b
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])
@property
def type_name(self) -> TypeName:
return "add"
def evaluate(self, a, b):
return a + b
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)]
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
def evaluate(self, a, b):
return a - b
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])
@property
def type_name(self) -> TypeName:
return "sub"
def evaluate(self, a, b):
return a - b
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)]
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
def evaluate(self, a, b):
return a * b
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])
@property
def type_name(self) -> TypeName:
return "mul"
def evaluate(self, a, b):
return a * b
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)]
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
def evaluate(self, a, b):
return a / b
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])
@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"
def evaluate(self, a, b):
return a / b
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)]
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return conjugate(a)
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])
@property
def type_name(self) -> TypeName:
return "conj"
return "min"
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,49 +126,49 @@ 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 = ""):
super().__init__(input_count = 2, output_count = 1, name = name, input_sources = [src0, src1])
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
@property
def type_name(self) -> 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
class SquareRoot(AbstractOperation):
"""Unary square root operation.
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])
@property
def type_name(self) -> TypeName:
return "max"
return "sqrt"
def evaluate(self, a):
return sqrt(complex(a))
class Min(AbstractOperation):
"""Binary min operation.
class ComplexConjugate(AbstractOperation):
"""Unary complex conjugate 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 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
def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 1, output_count = 1, name = name, input_sources = [src0])
@property
def type_name(self) -> TypeName:
return "min"
return "conj"
def evaluate(self, a):
return conjugate(a)
class Absolute(AbstractOperation):
......@@ -233,105 +176,71 @@ class Absolute(AbstractOperation):
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 np_abs(a)
def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 1, output_count = 1, name = name, input_sources = [src0])
@property
def type_name(self) -> TypeName:
return "abs"
def evaluate(self, a):
return np_abs(a)
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
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return a * self.param("coefficient")
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])
self.set_param("value", value)
@property
def type_name(self) -> TypeName:
return "cmul"
class ConstantAddition(AbstractOperation):
"""Unary constant addition 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
if source1 is not None:
self._input_ports[0].connect(source1)
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
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return a - self.param("coefficient")
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])
@property
def type_name(self) -> TypeName:
return "csub"
return "bfly"
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
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return a / self.param("coefficient")
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])
@property
def type_name(self) -> TypeName:
return "cdiv"
return "mad"
def evaluate(self, a, b, c):
return a * b + c
b_asic/graph_component.py
View file @ 58d9907d
......@@ -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):
......@@ -18,32 +23,87 @@ class GraphComponent(ABC):
@property
@abstractmethod
def type_name(self) -> TypeName:
"""Return the type name of the graph component"""
"""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 = {}
@property
def name(self) -> Name:
......@@ -52,3 +112,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)
\ No newline at end of file
b_asic/graph_id.py deleted 100644 → 0
View file @ 8ed17a9b
"""@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/operation.py
View file @ 58d9907d
This diff is collapsed.
b_asic/port.py
View file @ 58d9907d
......@@ -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 NewType, 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,33 @@ 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."""
raise NotImplementedError
@abstractmethod
def signal(self, i: int = 0) -> Signal:
"""Return the connected signal at index i.
Keyword argumens:
i: integer index of the signal requsted.
"""
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,22 +74,34 @@ class AbstractPort(Port):
Handles functionality for port id and saves the connection to the parent operation.
"""
_operation: "Operation"
_index: int
_operation: Operation
def __init__(self, index: int, operation: Operation):
self._index = index
def __init__(self, operation: "Operation", index: int):
self._operation = operation
self._index = index
@property
def operation(self) -> Operation:
def operation(self) -> "Operation":
return self._operation
@property
def index(self) -> PortIndex:
def index(self) -> int:
return self._index
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.
TODO: More info.
......@@ -120,104 +109,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 already removed signal."
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)
class OutputPort(AbstractPort):
@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
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 already removed signal."
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/signal.py
View file @ 58d9907d
"""@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)
@property
def type_name(self) -> 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,43 @@ 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)
\ No newline at end of file
b_asic/signal_flow_graph.py
View file @ 58d9907d
This diff is collapsed.
b_asic/simulation.py
View file @ 58d9907d
......@@ -3,33 +3,111 @@ 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 ResultKey, ResultMap
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, ResultMap]:
"""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)]
\ No newline at end of file
b_asic/special_operations.py 0 → 100644
View file @ 58d9907d
"""@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, ResultKey, RegisterMap, MutableResultMap, 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)
@property
def type_name(self) -> 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])
@property
def type_name(self) -> 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)
@property
def type_name(self) -> 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[MutableResultMap] = 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
\ No newline at end of file
small_logo.png 0 → 100644
View file @ 58d9907d
small_logo.png

39.5 KiB

src/main.cpp
View file @ 58d9907d
#include <pybind11/pybind11.h>
namespace py = pybind11;
namespace asic {
int add(int a, int b) {
return a + b;
}
int sub(int a, int b) {
return a - b;
}
} // namespace asic
PYBIND11_MODULE(_b_asic, m) {
m.doc() = "Better ASIC Toolbox Extension Module.";
m.def("add", &asic::add, "A function which adds two numbers.", py::arg("a"), py::arg("b"));
m.def("sub", &asic::sub, "A function which subtracts two numbers.", py::arg("a"), py::arg("b"));
#include <pybind11/pybind11.h>
namespace py = pybind11;
namespace asic {
int add(int a, int b) {
return a + b;
}
int sub(int a, int b) {
return a - b;
}
} // namespace asic
PYBIND11_MODULE(_b_asic, m) {
m.doc() = "Better ASIC Toolbox Extension Module.";
m.def("add", &asic::add, "A function which adds two numbers.", py::arg("a"), py::arg("b"));
m.def("sub", &asic::sub, "A function which subtracts two numbers.", py::arg("a"), py::arg("b"));
}
\ No newline at end of file
test/conftest.py
View file @ 58d9907d
from test.fixtures.signal import signal, signals
from test.fixtures.operation_tree import *
from test.fixtures.port import *
from test.fixtures.signal_flow_graph import *
import pytest
test/fixtures/operation_tree.py
View file @ 58d9907d
from b_asic.core_operations import Addition, Constant
from b_asic.signal import Signal
import pytest
from b_asic import Addition, Constant, Signal
@pytest.fixture
def operation():
return Constant(2)
def create_operation(_type, dest_oper, index, **kwargs):
oper = _type(**kwargs)
oper_signal = Signal()
oper._output_ports[0].add_signal(oper_signal)
dest_oper._input_ports[index].add_signal(oper_signal)
return oper
@pytest.fixture
def operation_tree():
"""Return a addition operation connected with 2 constants.
---C---+
---A
---C---+
"""Valid addition operation connected with 2 constants.
2---+
|
v
add = 2 + 3 = 5
^
|
3---+
"""
add_oper = Addition()
create_operation(Constant, add_oper, 0, value=2)
create_operation(Constant, add_oper, 1, value=3)
return add_oper
return Addition(Constant(2), Constant(3))
@pytest.fixture
def large_operation_tree():
"""Return a constant operation connected with a large operation tree with 3 other constants and 3 additions.
---C---+
---A---+
---C---+ |
+---A
---C---+ |
---A---+
---C---+
"""Valid addition operation connected with a large operation tree with 2 other additions and 4 constants.
2---+
|
v
add---+
^ |
| |
3---+ v
add = (2 + 3) + (4 + 5) = 14
4---+ ^
| |
v |
add---+
^
|
5---+
"""
add_oper = Addition()
add_oper_2 = Addition()
return Addition(Addition(Constant(2), Constant(3)), Addition(Constant(4), Constant(5)))
const_oper = create_operation(Constant, add_oper, 0, value=2)
create_operation(Constant, add_oper, 1, value=3)
create_operation(Constant, add_oper_2, 0, value=4)
create_operation(Constant, add_oper_2, 1, value=5)
add_oper_3 = Addition()
add_oper_signal = Signal(add_oper.output(0), add_oper_3.output(0))
add_oper._output_ports[0].add_signal(add_oper_signal)
add_oper_3._input_ports[0].add_signal(add_oper_signal)
add_oper_2_signal = Signal(add_oper_2.output(0), add_oper_3.output(0))
add_oper_2._output_ports[0].add_signal(add_oper_2_signal)
add_oper_3._input_ports[1].add_signal(add_oper_2_signal)
return const_oper
@pytest.fixture
def operation_graph_with_cycle():
"""Invalid addition operation connected with an operation graph containing a cycle.
+-+
| |
v |
add+---+
^ |
| v
7 add = (? + 7) + 6 = ?
^
|
6
"""
add1 = Addition(None, Constant(7))
add1.input(0).connect(add1)
return Addition(add1, Constant(6))
test/fixtures/port.py
View file @ 58d9907d
import pytest
from b_asic.port import InputPort, OutputPort
from b_asic import InputPort, OutputPort
@pytest.fixture
def input_port():
return InputPort(0, None)
return InputPort(None, 0)
@pytest.fixture
def output_port():
return OutputPort(0, None)
return OutputPort(None, 0)
@pytest.fixture
def list_of_input_ports():
return [InputPort(None, i) for i in range(0, 3)]
@pytest.fixture
def list_of_output_ports():
return [OutputPort(None, i) for i in range(0, 3)]
test/fixtures/signal.py
View file @ 58d9907d
import pytest
from b_asic import Signal
@pytest.fixture
def signal():
"""Return a signal with no connections."""
......@@ -9,4 +11,4 @@ def signal():
@pytest.fixture
def signals():
"""Return 3 signals with no connections."""
return [Signal() for _ in range(0,3)]
return [Signal() for _ in range(0, 3)]
test/fixtures/signal_flow_graph.py 0 → 100644
View file @ 58d9907d
import pytest
from b_asic import SFG, Input, Output, Constant, Register, ConstantMultiplication
@pytest.fixture
def sfg_two_inputs_two_outputs():
"""Valid SFG with two inputs and two outputs.
. .
in1-------+ +--------->out1
. | | .
. v | .
. add1+--+ .
. ^ | .
. | v .
in2+------+ add2---->out2
| . ^ .
| . | .
+------------+ .
. .
out1 = in1 + in2
out2 = in1 + 2 * in2
"""
in1 = Input()
in2 = Input()
add1 = in1 + in2
add2 = add1 + in2
out1 = Output(add1)
out2 = Output(add2)
return SFG(inputs = [in1, in2], outputs = [out1, out2])
@pytest.fixture
def sfg_nested():
"""Valid SFG with two inputs and one output.
out1 = in1 + (in1 + in1 * in2) * (in1 + in2 * (in1 + in1 * in2))
"""
mac_in1 = Input()
mac_in2 = Input()
mac_in3 = Input()
mac_out1 = Output(mac_in1 + mac_in2 * mac_in3)
MAC = SFG(inputs = [mac_in1, mac_in2, mac_in3], outputs = [mac_out1])
in1 = Input()
in2 = Input()
mac1 = MAC(in1, in1, in2)
mac2 = MAC(in1, in2, mac1)
mac3 = MAC(in1, mac1, mac2)
out1 = Output(mac3)
return SFG(inputs = [in1, in2], outputs = [out1])
@pytest.fixture
def sfg_delay():
"""Valid SFG with one input and one output.
out1 = in1'
"""
in1 = Input()
reg1 = Register(in1)
out1 = Output(reg1)
return SFG(inputs = [in1], outputs = [out1])
@pytest.fixture
def sfg_accumulator():
"""Valid SFG with two inputs and one output.
data_out = (data_in' + data_in) * (1 - reset)
"""
data_in = Input()
reset = Input()
reg = Register()
reg.input(0).connect((reg + data_in) * (1 - reset))
data_out = Output(reg)
return SFG(inputs = [data_in, reset], outputs = [data_out])
@pytest.fixture
def simple_filter():
"""A valid SFG that is used as a filter in the first lab for TSTE87.
+----<constmul1----+
| |
| |
in1>------add1>------reg>------+------out1>
"""
in1 = Input()
reg = Register()
constmul1 = ConstantMultiplication(0.5)
add1 = in1 + constmul1
reg.input(0).connect(add1)
constmul1.input(0).connect(reg)
out1 = Output(reg)
return SFG(inputs=[in1], outputs=[out1])
test/operation/test_abstract_operation.py → test/test_abstract_operation.py
View file @ 58d9907d
......@@ -2,11 +2,10 @@
B-ASIC test suite for the AbstractOperation class.
"""
from b_asic.core_operations import Addition, ConstantAddition, Subtraction, ConstantSubtraction, \
Multiplication, ConstantMultiplication, Division, ConstantDivision
import pytest
from b_asic import Addition, Subtraction, Multiplication, ConstantMultiplication, Division
def test_addition_overload():
"""Tests addition overloading for both operation and number argument."""
......@@ -14,15 +13,19 @@ def test_addition_overload():
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, ConstantAddition)
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():
......@@ -31,15 +34,19 @@ def test_subtraction_overload():
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, ConstantSubtraction)
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():
......@@ -48,15 +55,19 @@ def test_multiplication_overload():
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():
......@@ -65,13 +76,17 @@ def test_division_overload():
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, ConstantDivision)
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