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)
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
......@@ -3,266 +3,373 @@ B-ASIC Operation Module.
TODO: More info.
"""
import collections
from abc import abstractmethod
from numbers import Number
from typing import List, Dict, Optional, Any, Set, TYPE_CHECKING
from collections import deque
from typing import NewType, List, Sequence, Iterable, Mapping, MutableMapping, Optional, Any, Set, Union
from math import trunc
from b_asic.graph_component import GraphComponent, AbstractGraphComponent, Name
from b_asic.simulation import SimulationState, OperationState
from b_asic.port import SignalSourceProvider, InputPort, OutputPort
from b_asic.signal import Signal
if TYPE_CHECKING:
from b_asic.port import InputPort, OutputPort
ResultKey = NewType("ResultKey", str)
ResultMap = Mapping[ResultKey, Optional[Number]]
MutableResultMap = MutableMapping[ResultKey, Optional[Number]]
RegisterMap = Mapping[ResultKey, Number]
MutableRegisterMap = MutableMapping[ResultKey, Number]
class Operation(GraphComponent):
class Operation(GraphComponent, SignalSourceProvider):
"""Operation interface.
TODO: More info.
"""
@abstractmethod
def inputs(self) -> "List[InputPort]":
"""Get a list of all input ports."""
def __add__(self, src: Union[SignalSourceProvider, Number]) -> "Addition":
"""Overloads the addition operator to make it return a new Addition operation
object that is connected to the self and other objects.
"""
raise NotImplementedError
@abstractmethod
def __radd__(self, src: Union[SignalSourceProvider, Number]) -> "Addition":
"""Overloads the addition operator to make it return a new Addition operation
object that is connected to the self and other objects.
"""
raise NotImplementedError
@abstractmethod
def __sub__(self, src: Union[SignalSourceProvider, Number]) -> "Subtraction":
"""Overloads the subtraction operator to make it return a new Subtraction operation
object that is connected to the self and other objects.
"""
raise NotImplementedError
@abstractmethod
def __rsub__(self, src: Union[SignalSourceProvider, Number]) -> "Subtraction":
"""Overloads the subtraction operator to make it return a new Subtraction operation
object that is connected to the self and other objects.
"""
raise NotImplementedError
@abstractmethod
def outputs(self) -> "List[OutputPort]":
"""Get a list of all output ports."""
def __mul__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Multiplication, ConstantMultiplication]":
"""Overloads the multiplication operator to make it return a new Multiplication operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantMultiplication operation object instead.
"""
raise NotImplementedError
@abstractmethod
def __rmul__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Multiplication, ConstantMultiplication]":
"""Overloads the multiplication operator to make it return a new Multiplication operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantMultiplication operation object instead.
"""
raise NotImplementedError
@abstractmethod
def __truediv__(self, src: Union[SignalSourceProvider, Number]) -> "Division":
"""Overloads the division operator to make it return a new Division operation
object that is connected to the self and other objects.
"""
raise NotImplementedError
@abstractmethod
def __rtruediv__(self, src: Union[SignalSourceProvider, Number]) -> "Division":
"""Overloads the division operator to make it return a new Division operation
object that is connected to the self and other objects.
"""
raise NotImplementedError
@property
@abstractmethod
def input_count(self) -> int:
"""Get the number of input ports."""
raise NotImplementedError
@property
@abstractmethod
def output_count(self) -> int:
"""Get the number of output ports."""
raise NotImplementedError
@abstractmethod
def input(self, i: int) -> "InputPort":
"""Get the input port at index i."""
def input(self, index: int) -> InputPort:
"""Get the input port at the given index."""
raise NotImplementedError
@abstractmethod
def output(self, index: int) -> OutputPort:
"""Get the output port at the given index."""
raise NotImplementedError
@property
@abstractmethod
def output(self, i: int) -> "OutputPort":
"""Get the output port at index i."""
def inputs(self) -> Sequence[InputPort]:
"""Get all input ports."""
raise NotImplementedError
@property
@abstractmethod
def params(self) -> Dict[str, Optional[Any]]:
"""Get a dictionary of all parameter values."""
def outputs(self) -> Sequence[OutputPort]:
"""Get all output ports."""
raise NotImplementedError
@property
@abstractmethod
def param(self, name: str) -> Optional[Any]:
"""Get the value of a parameter.
Returns None if the parameter is not defined.
def input_signals(self) -> Iterable[Signal]:
"""Get all the signals that are connected to this operation's input ports,
in no particular order.
"""
raise NotImplementedError
@property
@abstractmethod
def set_param(self, name: str, value: Any) -> None:
"""Set the value of a parameter.
The parameter must be defined.
def output_signals(self) -> Iterable[Signal]:
"""Get all the signals that are connected to this operation's output ports,
in no particular order.
"""
raise NotImplementedError
@abstractmethod
def evaluate_outputs(self, state: "SimulationState") -> List[Number]:
"""Simulate the circuit until its iteration count matches that of the simulation state,
then return the resulting output vector.
def key(self, index: int, prefix: str = "") -> ResultKey:
"""Get the key used to access the result of a certain output of this operation
from the results parameter passed to current_output(s) or evaluate_output(s).
"""
raise NotImplementedError
@abstractmethod
def split(self) -> "List[Operation]":
"""Split the operation into multiple operations.
If splitting is not possible, this may return a list containing only the operation itself.
def current_output(self, index: int, registers: Optional[RegisterMap] = None, prefix: str = "") -> Optional[Number]:
"""Get the current output at the given index of this operation, if available.
The registers parameter will be used for lookup.
The prefix parameter will be used as a prefix for the key string when looking for registers.
See also: current_outputs, evaluate_output, evaluate_outputs.
"""
raise NotImplementedError
@property
@abstractmethod
def neighbors(self) -> "List[Operation]":
"""Return all operations that are connected by signals to this operation.
If no neighbors are found, this returns an empty list.
def evaluate_output(self, index: int, input_values: Sequence[Number], results: Optional[MutableResultMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = "") -> Number:
"""Evaluate the output at the given index of this operation with the given input values.
The results parameter will be used to store any results (including intermediate results) for caching.
The registers parameter will be used to get the current value of any intermediate registers that are encountered, and be updated with their new values.
The prefix parameter will be used as a prefix for the key string when storing results/registers.
See also: evaluate_outputs, current_output, current_outputs.
"""
raise NotImplementedError
@abstractmethod
def current_outputs(self, registers: Optional[RegisterMap] = None, prefix: str = "") -> Sequence[Optional[Number]]:
"""Get all current outputs of this operation, if available.
See current_output for more information.
"""
raise NotImplementedError
@abstractmethod
def evaluate_outputs(self, input_values: Sequence[Number], results: Optional[MutableResultMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = "") -> Sequence[Number]:
"""Evaluate all outputs of this operation given the input values.
See evaluate_output for more information.
"""
raise NotImplementedError
@abstractmethod
def split(self) -> Iterable["Operation"]:
"""Split the operation into multiple operations.
If splitting is not possible, this may return a list containing only the operation itself.
"""
raise NotImplementedError
class AbstractOperation(Operation, AbstractGraphComponent):
"""Generic abstract operation class which most implementations will derive from.
TODO: More info.
"""
_input_ports: List["InputPort"]
_output_ports: List["OutputPort"]
_parameters: Dict[str, Optional[Any]]
_input_ports: List[InputPort]
_output_ports: List[OutputPort]
def __init__(self, name: Name = ""):
def __init__(self, input_count: int, output_count: int, name: Name = "", input_sources: Optional[Sequence[Optional[SignalSourceProvider]]] = None):
super().__init__(name)
self._input_ports = []
self._output_ports = []
self._parameters = {}
self._input_ports = [InputPort(self, i) for i in range(input_count)] # Allocate input ports.
self._output_ports = [OutputPort(self, i) for i in range(output_count)] # Allocate output ports.
# Connect given input sources, if any.
if input_sources is not None:
source_count = len(input_sources)
if source_count != input_count:
raise ValueError(f"Wrong number of input sources supplied to Operation (expected {input_count}, got {source_count})")
for i, src in enumerate(input_sources):
if src is not None:
self._input_ports[i].connect(src.source)
@abstractmethod
def evaluate(self, *inputs) -> Any: # pylint: disable=arguments-differ
"""Evaluate the operation and generate a list of output values given a
list of input values.
"""
def evaluate(self, *inputs) -> Any: # pylint: disable=arguments-differ
"""Evaluate the operation and generate a list of output values given a list of input values."""
raise NotImplementedError
def inputs(self) -> List["InputPort"]:
return self._input_ports.copy()
def outputs(self) -> List["OutputPort"]:
return self._output_ports.copy()
def __add__(self, src: Union[SignalSourceProvider, Number]) -> "Addition":
from b_asic.core_operations import Constant, Addition # Import here to avoid circular imports.
return Addition(self, Constant(src) if isinstance(src, Number) else src)
def __radd__(self, src: Union[SignalSourceProvider, Number]) -> "Addition":
from b_asic.core_operations import Constant, Addition # Import here to avoid circular imports.
return Addition(Constant(src) if isinstance(src, Number) else src, self)
def __sub__(self, src: Union[SignalSourceProvider, Number]) -> "Subtraction":
from b_asic.core_operations import Constant, Subtraction # Import here to avoid circular imports.
return Subtraction(self, Constant(src) if isinstance(src, Number) else src)
def __rsub__(self, src: Union[SignalSourceProvider, Number]) -> "Subtraction":
from b_asic.core_operations import Constant, Subtraction # Import here to avoid circular imports.
return Subtraction(Constant(src) if isinstance(src, Number) else src, self)
def __mul__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Multiplication, ConstantMultiplication]":
from b_asic.core_operations import Multiplication, ConstantMultiplication # Import here to avoid circular imports.
return ConstantMultiplication(src, self) if isinstance(src, Number) else Multiplication(self, src)
def __rmul__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Multiplication, ConstantMultiplication]":
from b_asic.core_operations import Multiplication, ConstantMultiplication # Import here to avoid circular imports.
return ConstantMultiplication(src, self) if isinstance(src, Number) else Multiplication(src, self)
def __truediv__(self, src: Union[SignalSourceProvider, Number]) -> "Division":
from b_asic.core_operations import Constant, Division # Import here to avoid circular imports.
return Division(self, Constant(src) if isinstance(src, Number) else src)
def __rtruediv__(self, src: Union[SignalSourceProvider, Number]) -> "Division":
from b_asic.core_operations import Constant, Division # Import here to avoid circular imports.
return Division(Constant(src) if isinstance(src, Number) else src, self)
@property
def input_count(self) -> int:
return len(self._input_ports)
@property
def output_count(self) -> int:
return len(self._output_ports)
def input(self, i: int) -> "InputPort":
return self._input_ports[i]
def output(self, i: int) -> "OutputPort":
return self._output_ports[i]
def params(self) -> Dict[str, Optional[Any]]:
return self._parameters.copy()
def param(self, name: str) -> Optional[Any]:
return self._parameters.get(name)
def set_param(self, name: str, value: Any) -> None:
assert name in self._parameters # TODO: Error message.
self._parameters[name] = value
def evaluate_outputs(self, state: SimulationState) -> List[Number]:
# TODO: Check implementation.
input_count: int = self.input_count()
output_count: int = self.output_count()
assert input_count == len(self._input_ports) # TODO: Error message.
assert output_count == len(self._output_ports) # TODO: Error message.
self_state: OperationState = state.operation_states[self]
def input(self, index: int) -> InputPort:
return self._input_ports[index]
while self_state.iteration < state.iteration:
input_values: List[Number] = [0] * input_count
for i in range(input_count):
source: Signal = self._input_ports[i].signal
input_values[i] = source.operation.evaluate_outputs(state)[
source.port_index]
def output(self, index: int) -> OutputPort:
return self._output_ports[index]
self_state.output_values = self.evaluate(input_values)
# TODO: Error message.
assert len(self_state.output_values) == output_count
self_state.iteration += 1
for i in range(output_count):
for signal in self._output_ports[i].signals():
destination: Signal = signal.destination
destination.evaluate_outputs(state)
return self_state.output_values
@property
def inputs(self) -> Sequence[InputPort]:
return self._input_ports
def split(self) -> List[Operation]:
# TODO: Check implementation.
results = self.evaluate(self._input_ports)
if all(isinstance(e, Operation) for e in results):
return results
return [self]
@property
def outputs(self) -> Sequence[OutputPort]:
return self._output_ports
@property
def neighbors(self) -> List[Operation]:
neighbors: List[Operation] = []
for port in self._input_ports:
for signal in port.signals:
neighbors.append(signal.source.operation)
for port in self._output_ports:
for signal in port.signals:
neighbors.append(signal.destination.operation)
return neighbors
def traverse(self) -> Operation:
"""Traverse the operation tree and return a generator with start point in the operation."""
return self._breadth_first_search()
def _breadth_first_search(self) -> Operation:
"""Use breadth first search to traverse the operation tree."""
visited: Set[Operation] = {self}
queue = deque([self])
while queue:
operation = queue.popleft()
yield operation
for n_operation in operation.neighbors:
if n_operation not in visited:
visited.add(n_operation)
queue.append(n_operation)
def __add__(self, other):
"""Overloads the addition operator to make it return a new Addition operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantAddition operation object instead.
"""
# Import here to avoid circular imports.
from b_asic.core_operations import Addition, ConstantAddition
def input_signals(self) -> Iterable[Signal]:
result = []
for p in self.inputs:
for s in p.signals:
result.append(s)
return result
if isinstance(other, Operation):
return Addition(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantAddition(other, self.output(0))
@property
def output_signals(self) -> Iterable[Signal]:
result = []
for p in self.outputs:
for s in p.signals:
result.append(s)
return result
def key(self, index: int, prefix: str = "") -> ResultKey:
key = prefix
if self.output_count != 1:
if key:
key += "."
key += str(index)
elif not key:
key = str(index)
return key
def current_output(self, index: int, registers: Optional[RegisterMap] = None, prefix: str = "") -> Optional[Number]:
return None
def evaluate_output(self, index: int, input_values: Sequence[Number], results: Optional[MutableResultMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = "") -> Number:
if index < 0 or index >= self.output_count:
raise IndexError(f"Output index out of range (expected 0-{self.output_count - 1}, got {index})")
if len(input_values) != self.input_count:
raise ValueError(f"Wrong number of input values supplied to operation (expected {self.input_count}, got {len(input_values)})")
if results is None:
results = {}
if registers is None:
registers = {}
values = self.evaluate(*self.truncate_inputs(input_values))
if isinstance(values, collections.abc.Sequence):
if len(values) != self.output_count:
raise RuntimeError(f"Operation evaluated to incorrect number of outputs (expected {self.output_count}, got {len(values)})")
elif isinstance(values, Number):
if self.output_count != 1:
raise RuntimeError(f"Operation evaluated to incorrect number of outputs (expected {self.output_count}, got 1)")
values = (values,)
else:
raise TypeError("Other type is not an Operation or a Number.")
def __sub__(self, other):
"""Overloads the subtraction operator to make it return a new Subtraction operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantSubtraction operation object instead.
"""
# Import here to avoid circular imports.
from b_asic.core_operations import Subtraction, ConstantSubtraction
raise RuntimeError(f"Operation evaluated to invalid type (expected Sequence/Number, got {values.__class__.__name__})")
if isinstance(other, Operation):
return Subtraction(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantSubtraction(other, self.output(0))
if self.output_count == 1:
results[self.key(index, prefix)] = values[index]
else:
raise TypeError("Other type is not an Operation or a Number.")
for i in range(self.output_count):
results[self.key(i, prefix)] = values[i]
return values[index]
def __mul__(self, other):
"""Overloads the multiplication operator to make it return a new Multiplication operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantMultiplication operation object instead.
"""
# Import here to avoid circular imports.
from b_asic.core_operations import Multiplication, ConstantMultiplication
def current_outputs(self, registers: Optional[RegisterMap] = None, prefix: str = "") -> Sequence[Optional[Number]]:
return [self.current_output(i, registers, prefix) for i in range(self.output_count)]
if isinstance(other, Operation):
return Multiplication(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantMultiplication(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")
def evaluate_outputs(self, input_values: Sequence[Number], results: Optional[MutableResultMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = "") -> Sequence[Number]:
return [self.evaluate_output(i, input_values, results, registers, prefix) for i in range(self.output_count)]
def __truediv__(self, other):
"""Overloads the division operator to make it return a new Division operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantDivision operation object instead.
"""
def split(self) -> Iterable[Operation]:
# Import here to avoid circular imports.
from b_asic.core_operations import Division, ConstantDivision
from b_asic.special_operations import Input
try:
result = self.evaluate(*[Input()] * self.input_count)
if isinstance(result, collections.Sequence) and all(isinstance(e, Operation) for e in result):
return result
if isinstance(result, Operation):
return [result]
except TypeError:
pass
except ValueError:
pass
return [self]
if isinstance(other, Operation):
return Division(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantDivision(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")
@property
def neighbors(self) -> Iterable[GraphComponent]:
return list(self.input_signals) + list(self.output_signals)
@property
def source(self) -> OutputPort:
if self.output_count != 1:
diff = "more" if self.output_count > 1 else "less"
raise TypeError(
f"{self.__class__.__name__} cannot be used as an input source because it has {diff} than 1 output")
return self.output(0)
def truncate_input(self, index: int, value: Number, bits: int) -> Number:
"""Truncate the value to be used as input at the given index to a certain bit length."""
n = value
if not isinstance(n, int):
n = trunc(value)
return n & ((2 ** bits) - 1)
def truncate_inputs(self, input_values: Sequence[Number]) -> Sequence[Number]:
"""Truncate the values to be used as inputs to the bit lengths specified by the respective signals connected to each input."""
args = []
for i, input_port in enumerate(self.inputs):
if input_port.signal_count >= 1:
bits = input_port.signals[0].bits
if bits is None:
args.append(input_values[i])
else:
if isinstance(input_values[i], complex):
raise TypeError("Complex value cannot be truncated to {bits} bits as requested by the signal connected to input #{i}")
args.append(self.truncate_input(i, input_values[i], bits))
else:
args.append(input_values[i])
return args
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
......@@ -3,14 +3,34 @@ B-ASIC Signal Flow Graph Module.
TODO: More info.
"""
from typing import List, Dict, Optional, DefaultDict
from collections import defaultdict
from typing import List, Iterable, Sequence, Dict, Optional, DefaultDict, Set
from numbers import Number
from collections import defaultdict, deque
from b_asic.operation import Operation
from b_asic.operation import AbstractOperation
from b_asic.port import SignalSourceProvider, OutputPort
from b_asic.operation import Operation, AbstractOperation, ResultKey, RegisterMap, MutableResultMap, MutableRegisterMap
from b_asic.signal import Signal
from b_asic.graph_id import GraphIDGenerator, GraphID
from b_asic.graph_component import GraphComponent, Name, TypeName
from b_asic.graph_component import GraphID, GraphIDNumber, GraphComponent, Name, TypeName
from b_asic.special_operations import Input, Output
class GraphIDGenerator:
"""A class that generates Graph IDs for objects."""
_next_id_number: DefaultDict[TypeName, GraphIDNumber]
def __init__(self, id_number_offset: GraphIDNumber = 0):
self._next_id_number = defaultdict(lambda: id_number_offset)
def next_id(self, type_name: TypeName) -> GraphID:
"""Get the next graph id for a certain graph id type."""
self._next_id_number[type_name] += 1
return type_name + str(self._next_id_number[type_name])
@property
def id_number_offset(self) -> GraphIDNumber:
"""Get the graph id number offset of this generator."""
return self._next_id_number.default_factory() # pylint: disable=not-callable
class SFG(AbstractOperation):
......@@ -18,74 +38,419 @@ class SFG(AbstractOperation):
TODO: More info.
"""
_graph_components_by_id: Dict[GraphID, GraphComponent]
_graph_components_by_name: DefaultDict[Name, List[GraphComponent]]
_components_by_id: Dict[GraphID, GraphComponent]
_components_by_name: DefaultDict[Name, List[GraphComponent]]
_components_ordered: List[GraphComponent]
_operations_ordered: List[Operation]
_graph_id_generator: GraphIDGenerator
_input_operations: List[Input]
_output_operations: List[Output]
_original_components_to_new: Set[GraphComponent]
_original_input_signals_to_indices: Dict[Signal, int]
_original_output_signals_to_indices: Dict[Signal, int]
def __init__(self, input_signals: List[Signal] = None, output_signals: List[Signal] = None, \
ops: List[Operation] = None, **kwds):
super().__init__(**kwds)
if input_signals is None:
input_signals = []
if output_signals is None:
output_signals = []
if ops is None:
ops = []
def __init__(self, input_signals: Optional[Sequence[Signal]] = None, output_signals: Optional[Sequence[Signal]] = None, \
inputs: Optional[Sequence[Input]] = None, outputs: Optional[Sequence[Output]] = None, \
id_number_offset: GraphIDNumber = 0, name: Name = "", \
input_sources: Optional[Sequence[Optional[SignalSourceProvider]]] = None):
input_signal_count = 0 if input_signals is None else len(input_signals)
input_operation_count = 0 if inputs is None else len(inputs)
output_signal_count = 0 if output_signals is None else len(output_signals)
output_operation_count = 0 if outputs is None else len(outputs)
super().__init__(input_count = input_signal_count + input_operation_count,
output_count = output_signal_count + output_operation_count,
name = name, input_sources = input_sources)
self._graph_components_by_id = dict() # Maps Graph ID to objects
self._graph_components_by_name = defaultdict(list) # Maps Name to objects
self._graph_id_generator = GraphIDGenerator()
self._components_by_id = dict()
self._components_by_name = defaultdict(list)
self._components_ordered = []
self._operations_ordered = []
self._graph_id_generator = GraphIDGenerator(id_number_offset)
self._input_operations = []
self._output_operations = []
self._original_components_to_new = {}
self._original_input_signals_to_indices = {}
self._original_output_signals_to_indices = {}
for operation in ops:
self._add_graph_component(operation)
# Setup input signals.
if input_signals is not None:
for input_index, signal in enumerate(input_signals):
assert signal not in self._original_components_to_new, "Duplicate input signals supplied to SFG construcctor."
new_input_op = self._add_component_unconnected_copy(Input())
new_signal = self._add_component_unconnected_copy(signal)
new_signal.set_source(new_input_op.output(0))
self._input_operations.append(new_input_op)
self._original_input_signals_to_indices[signal] = input_index
for input_signal in input_signals:
self._add_graph_component(input_signal)
# Setup input operations, starting from indices ater input signals.
if inputs is not None:
for input_index, input_op in enumerate(inputs, input_signal_count):
assert input_op not in self._original_components_to_new, "Duplicate input operations supplied to SFG constructor."
new_input_op = self._add_component_unconnected_copy(input_op)
for signal in input_op.output(0).signals:
assert signal not in self._original_components_to_new, "Duplicate input signals connected to input ports supplied to SFG construcctor."
new_signal = self._add_component_unconnected_copy(signal)
new_signal.set_source(new_input_op.output(0))
self._original_input_signals_to_indices[signal] = input_index
# TODO: Construct SFG based on what inputs that were given
# TODO: Traverse the graph between the inputs/outputs and add to self._operations.
# TODO: Connect ports with signals with appropriate IDs.
self._input_operations.append(new_input_op)
def evaluate(self, *inputs) -> list:
return [] # TODO: Implement
# Setup output signals.
if output_signals is not None:
for output_index, signal in enumerate(output_signals):
new_output_op = self._add_component_unconnected_copy(Output())
if signal in self._original_components_to_new:
# Signal was already added when setting up inputs.
new_signal = self._original_components_to_new[signal]
new_signal.set_destination(new_output_op.input(0))
else:
# New signal has to be created.
new_signal = self._add_component_unconnected_copy(signal)
new_signal.set_destination(new_output_op.input(0))
def _add_graph_component(self, graph_component: GraphComponent) -> GraphID:
"""Add the entered graph component to the SFG's dictionary of graph objects and
return a generated GraphID for it.
self._output_operations.append(new_output_op)
self._original_output_signals_to_indices[signal] = output_index
Keyword arguments:
graph_component: Graph component to add to the graph.
"""
# Add to name dict
self._graph_components_by_name[graph_component.name].append(graph_component)
# Setup output operations, starting from indices after output signals.
if outputs is not None:
for output_index, output_op in enumerate(outputs, output_signal_count):
assert output_op not in self._original_components_to_new, "Duplicate output operations supplied to SFG constructor."
new_output_op = self._add_component_unconnected_copy(output_op)
for signal in output_op.input(0).signals:
new_signal = None
if signal in self._original_components_to_new:
# Signal was already added when setting up inputs.
new_signal = self._original_components_to_new[signal]
else:
# New signal has to be created.
new_signal = self._add_component_unconnected_copy(signal)
new_signal.set_destination(new_output_op.input(0))
self._original_output_signals_to_indices[signal] = output_index
self._output_operations.append(new_output_op)
output_operations_set = set(self._output_operations)
# Search the graph inwards from each input signal.
for signal, input_index in self._original_input_signals_to_indices.items():
# Check if already added destination.
new_signal = self._original_components_to_new[signal]
if new_signal.destination is None:
if signal.destination is None:
raise ValueError(f"Input signal #{input_index} is missing destination in SFG")
if signal.destination.operation not in self._original_components_to_new:
self._add_operation_connected_tree_copy(signal.destination.operation)
elif new_signal.destination.operation in output_operations_set:
# Add directly connected input to output to ordered list.
self._components_ordered.extend([new_signal.source.operation, new_signal, new_signal.destination.operation])
self._operations_ordered.extend([new_signal.source.operation, new_signal.destination.operation])
# Search the graph inwards from each output signal.
for signal, output_index in self._original_output_signals_to_indices.items():
# Check if already added source.
new_signal = self._original_components_to_new[signal]
if new_signal.source is None:
if signal.source is None:
raise ValueError(f"Output signal #{output_index} is missing source in SFG")
if signal.source.operation not in self._original_components_to_new:
self._add_operation_connected_tree_copy(signal.source.operation)
def __str__(self) -> str:
"""Get a string representation of this SFG."""
output_string = ""
for component in self._components_ordered:
if isinstance(component, Operation):
for key, value in self._components_by_id.items():
if value is component:
output_string += "id: " + key + ", name: "
if component.name != None:
output_string += component.name + ", "
else:
output_string += "-, "
if component.type_name is "c":
output_string += "value: " + str(component.value) + ", input: ["
else:
output_string += "input: ["
counter_input = 0
for input in component.inputs:
counter_input += 1
for signal in input.signals:
for key, value in self._components_by_id.items():
if value is signal:
output_string += key + ", "
if counter_input > 0:
output_string = output_string[:-2]
output_string += "], output: ["
counter_output = 0
for output in component.outputs:
counter_output += 1
for signal in output.signals:
for key, value in self._components_by_id.items():
if value is signal:
output_string += key + ", "
if counter_output > 0:
output_string = output_string[:-2]
output_string += "]\n"
return output_string
# Add to ID dict
graph_id: GraphID = self._graph_id_generator.get_next_id(graph_component.type_name)
self._graph_components_by_id[graph_id] = graph_component
return graph_id
def __call__(self, *src: Optional[SignalSourceProvider], name: Name = "") -> "SFG":
"""Get a new independent SFG instance that is identical to this SFG except without any of its external connections."""
return SFG(inputs = self._input_operations, outputs = self._output_operations,
id_number_offset = self.id_number_offset, name = name, input_sources = src if src else None)
@property
def type_name(self) -> TypeName:
return "sfg"
def evaluate(self, *args):
result = self.evaluate_outputs(args, {}, {}, "")
n = len(result)
return None if n == 0 else result[0] if n == 1 else result
def evaluate_output(self, index: int, input_values: Sequence[Number], results: Optional[MutableResultMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = "") -> Number:
if index < 0 or index >= self.output_count:
raise IndexError(f"Output index out of range (expected 0-{self.output_count - 1}, got {index})")
if len(input_values) != self.input_count:
raise ValueError(f"Wrong number of inputs supplied to SFG for evaluation (expected {self.input_count}, got {len(input_values)})")
if results is None:
results = {}
if registers is None:
registers = {}
# Set the values of our input operations to the given input values.
for op, arg in zip(self._input_operations, self.truncate_inputs(input_values)):
op.value = arg
value = self._evaluate_source(self._output_operations[index].input(0).signals[0].source, results, registers, prefix)
results[self.key(index, prefix)] = value
return value
def split(self) -> Iterable[Operation]:
return self.operations
def copy_component(self, *args, **kwargs) -> GraphComponent:
return super().copy_component(*args, **kwargs, inputs = self._input_operations, outputs = self._output_operations,
id_number_offset = self.id_number_offset, name = self.name)
@property
def id_number_offset(self) -> GraphIDNumber:
"""Get the graph id number offset of the graph id generator for this SFG."""
return self._graph_id_generator.id_number_offset
@property
def components(self) -> Iterable[GraphComponent]:
"""Get all components of this graph in depth-first order."""
return self._components_ordered
@property
def operations(self) -> Iterable[Operation]:
"""Get all operations of this graph in depth-first order."""
return self._operations_ordered
def find_by_id(self, graph_id: GraphID) -> Optional[GraphComponent]:
"""Find a graph object based on the entered Graph ID and return it. If no graph
object with the entered ID was found then return None.
"""Find the graph component with the specified ID.
Returns None if the component was not found.
Keyword arguments:
graph_id: Graph ID of the desired component(s)
"""
return self._components_by_id.get(graph_id, None)
def find_by_name(self, name: Name) -> Sequence[GraphComponent]:
"""Find all graph components with the specified name.
Returns an empty sequence if no components were found.
Keyword arguments:
graph_id: Graph ID of the wanted object.
name: Name of the desired component(s)
"""
if graph_id in self._graph_components_by_id:
return self._graph_components_by_id[graph_id]
return self._components_by_name.get(name, [])
def _add_component_unconnected_copy(self, original_component: GraphComponent) -> GraphComponent:
assert original_component not in self._original_components_to_new, "Tried to add duplicate SFG component"
new_component = original_component.copy_component()
self._original_components_to_new[original_component] = new_component
new_id = self._graph_id_generator.next_id(new_component.type_name)
new_component.graph_id = new_id
self._components_by_id[new_id] = new_component
self._components_by_name[new_component.name].append(new_component)
return new_component
def _add_operation_connected_tree_copy(self, start_op: Operation) -> None:
op_stack = deque([start_op])
while op_stack:
original_op = op_stack.pop()
# Add or get the new copy of the operation.
new_op = None
if original_op not in self._original_components_to_new:
new_op = self._add_component_unconnected_copy(original_op)
self._components_ordered.append(new_op)
self._operations_ordered.append(new_op)
else:
new_op = self._original_components_to_new[original_op]
# Connect input ports to new signals.
for original_input_port in original_op.inputs:
if original_input_port.signal_count < 1:
raise ValueError("Unconnected input port in SFG")
for original_signal in original_input_port.signals:
# Check if the signal is one of the SFG's input signals.
if original_signal in self._original_input_signals_to_indices:
# New signal already created during first step of constructor.
new_signal = self._original_components_to_new[original_signal]
new_signal.set_destination(new_op.input(original_input_port.index))
self._components_ordered.extend([new_signal, new_signal.source.operation])
self._operations_ordered.append(new_signal.source.operation)
# Check if the signal has not been added before.
elif original_signal not in self._original_components_to_new:
if original_signal.source is None:
raise ValueError("Dangling signal without source in SFG")
new_signal = self._add_component_unconnected_copy(original_signal)
new_signal.set_destination(new_op.input(original_input_port.index))
self._components_ordered.append(new_signal)
original_connected_op = original_signal.source.operation
# Check if connected Operation has been added before.
if original_connected_op in self._original_components_to_new:
# Set source to the already added operations port.
new_signal.set_source(self._original_components_to_new[original_connected_op].output(original_signal.source.index))
else:
# Create new operation, set signal source to it.
new_connected_op = self._add_component_unconnected_copy(original_connected_op)
new_signal.set_source(new_connected_op.output(original_signal.source.index))
self._components_ordered.append(new_connected_op)
self._operations_ordered.append(new_connected_op)
# Add connected operation to queue of operations to visit.
op_stack.append(original_connected_op)
return None
# Connect output ports.
for original_output_port in original_op.outputs:
for original_signal in original_output_port.signals:
# Check if the signal is one of the SFG's output signals.
if original_signal in self._original_output_signals_to_indices:
# New signal already created during first step of constructor.
new_signal = self._original_components_to_new[original_signal]
new_signal.set_source(new_op.output(original_output_port.index))
self._components_ordered.extend([new_signal, new_signal.destination.operation])
self._operations_ordered.append(new_signal.destination.operation)
def find_by_name(self, name: Name) -> List[GraphComponent]:
"""Find all graph objects that have the entered name and return them
in a list. If no graph object with the entered name was found then return an
empty list.
# Check if signal has not been added before.
elif original_signal not in self._original_components_to_new:
if original_signal.source is None:
raise ValueError("Dangling signal without source in SFG")
new_signal = self._add_component_unconnected_copy(original_signal)
new_signal.set_source(new_op.output(original_output_port.index))
self._components_ordered.append(new_signal)
original_connected_op = original_signal.destination.operation
# Check if connected operation has been added.
if original_connected_op in self._original_components_to_new:
# Set destination to the already connected operations port.
new_signal.set_destination(self._original_components_to_new[original_connected_op].input(original_signal.destination.index))
else:
# Create new operation, set destination to it.
new_connected_op = self._add_component_unconnected_copy(original_connected_op)
new_signal.set_destination(new_connected_op.input(original_signal.destination.index))
self._components_ordered.append(new_connected_op)
self._operations_ordered.append(new_connected_op)
# Add connected operation to the queue of operations to visit.
op_stack.append(original_connected_op)
def replace_component(self, component: Operation, _component: Operation = None, _id: GraphID = None):
"""Find and replace all components matching either on GraphID, Type or both.
Then return a new deepcopy of the sfg with the replaced component.
Arguments:
component: The new component(s), e.g Multiplication
Keyword arguments:
name: Name of the wanted object.
_component: The specific component to replace.
_id: The GraphID to match the component to replace.
"""
return self._graph_components_by_name[name]
@property
def type_name(self) -> TypeName:
return "sfg"
assert _component is not None or _id is not None, \
"Define either operation to replace or GraphID of operation"
if _id is not None:
_component = self.find_by_id(_id)
assert _component is not None and isinstance(_component, Operation), \
"No operation matching the criteria found"
assert _component.output_count == component.output_count, \
"The output count may not differ between the operations"
assert _component.input_count == component.input_count, \
"The input count may not differ between the operations"
for index_in, _inp in enumerate(_component.inputs):
for _signal in _inp.signals:
_signal.remove_destination()
_signal.set_destination(component.input(index_in))
for index_out, _out in enumerate(_component.outputs):
for _signal in _out.signals:
_signal.remove_source()
_signal.set_source(component.output(index_out))
# The old SFG will be deleted by Python GC
return self()
def replace_operations(self, operation_ids: Sequence[GraphID], operation: Operation):
"""Replace multiple operations in the sfg with a operation with the same amount of inputs and outputs.
Then return a new deepcopy of the sfg with the replaced operations.
Arguments:
operation_ids: The id's of the operations we are replacing
operation: The operation used for replacement.
"""
_sfg = self()
input_signals = []
output_signals = []
for _operation in [_sfg.find_by_id(_id) for _id in operation_ids]:
input_signals.extend(filter(lambda s: s.source.operation.graph_id not in operation_ids, _operation.input_signals))
output_signals.extend(filter(lambda s: s.destination.operation.graph_id not in operation_ids, _operation.output_signals))
assert len(input_signals) == operation.input_count, "The input count must match"
assert len(output_signals) == operation.output_count, "The output count must match"
for index_in, _signal in enumerate(input_signals):
_signal.remove_destination()
_signal.set_destination(operation.input(index_in))
for index_out, _signal in enumerate(output_signals):
_signal.remove_source()
_signal.set_source(operation.output(index_out))
return _sfg()
def _evaluate_source(self, src: OutputPort, results: MutableResultMap, registers: MutableRegisterMap, prefix: str) -> Number:
src_prefix = prefix
if src_prefix:
src_prefix += "."
src_prefix += src.operation.graph_id
key = src.operation.key(src.index, src_prefix)
if key in results:
value = results[key]
if value is None:
raise RuntimeError(f"Direct feedback loop detected when evaluating operation.")
return value
results[key] = src.operation.current_output(src.index, registers, src_prefix)
input_values = [self._evaluate_source(input_port.signals[0].source, results, registers, prefix) for input_port in src.operation.inputs]
value = src.operation.evaluate_output(src.index, input_values, results, registers, src_prefix)
results[key] = value
return value
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