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
Select Git revision

Target

Select target project
  • da/B-ASIC
  • lukja239/B-ASIC
  • robal695/B-ASIC
3 results
Select Git revision
Show changes
Hide whitespace changes
Inline Side-by-side
Showing
with 1614 additions and 747 deletions
b_asic/GUI/operation_icons/sub_grey.png 0 → 100644
View file @ 5ca593d6
b_asic/GUI/operation_icons/sub_grey.png

1.18 KiB

b_asic/GUI/port_button.py 0 → 100644
View file @ 5ca593d6
import sys
from PyQt5.QtWidgets import QPushButton, QMenu
from PyQt5.QtCore import Qt, pyqtSignal
class PortButton(QPushButton):
connectionRequested = pyqtSignal(QPushButton)
moved = pyqtSignal()
def __init__(self, name, operation, window, parent=None):
self.pressed = False
self.window = window
self.operation = operation
self.clicked = 0
super(PortButton, self).__init__(name, operation)
def contextMenuEvent(self, event):
menu = QMenu()
menu.addAction("Connect", lambda: self.connectionRequested.emit(self))
menu.exec_(self.cursor().pos())
def mousePressEvent(self, event):
if event.button() == Qt.LeftButton:
self.clicked += 1
if self.clicked == 1:
self.setStyleSheet("background-color: grey")
self.pressed = True
self.window.pressed_ports.append(self)
elif self.clicked == 2:
self.setStyleSheet("background-color: white")
self.pressed = False
self.clicked = 0
self.window.pressed_ports.remove(self)
super(PortButton, self).mousePressEvent(event)
def mouseReleaseEvent(self, event):
super(PortButton, self).mouseReleaseEvent(event)
b_asic/GUI/utils.py 0 → 100644
View file @ 5ca593d6
from PyQt5.QtWidgets import QErrorMessage
from traceback import format_exc
def handle_error(fn):
def wrapper(self, *args, **kwargs):
try:
return fn(self, *args, **kwargs)
except Exception as e:
QErrorMessage(self._window).showMessage(f"Unexpected error: {format_exc()}")
return wrapper
def decorate_class(decorator):
def decorate(cls):
for attr in cls.__dict__:
if callable(getattr(cls, attr)):
setattr(cls, attr, decorator(getattr(cls, attr)))
return cls
return decorate
\ No newline at end of file
b_asic/__init__.py
View file @ 5ca593d6
...@@ -4,11 +4,9 @@ TODO: More info. ...@@ -4,11 +4,9 @@ TODO: More info.
""" """
from b_asic.core_operations import * from b_asic.core_operations import *
from b_asic.graph_component import * from b_asic.graph_component import *
from b_asic.graph_id import *
from b_asic.operation import * from b_asic.operation import *
from b_asic.precedence_chart import *
from b_asic.port import * from b_asic.port import *
from b_asic.schema import *
from b_asic.signal_flow_graph import * from b_asic.signal_flow_graph import *
from b_asic.signal import * from b_asic.signal import *
from b_asic.simulation import * from b_asic.simulation import *
from b_asic.special_operations import *
b_asic/core_operations.py
View file @ 5ca593d6
...@@ -4,43 +4,39 @@ TODO: More info. ...@@ -4,43 +4,39 @@ TODO: More info.
""" """
from numbers import Number from numbers import Number
from typing import Any from typing import Optional
from numpy import conjugate, sqrt, abs as np_abs 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.operation import AbstractOperation
from b_asic.graph_component import Name, TypeName 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): class Constant(AbstractOperation):
"""Constant value operation. """Constant value operation.
TODO: More info. TODO: More info.
""" """
def __init__(self, value: Number = 0, name: Name = ""): 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)] @classmethod
self._parameters["value"] = value def type_name(cls) -> TypeName:
return "c"
def evaluate(self): def evaluate(self):
return self.param("value") return self.param("value")
@property @property
def type_name(self) -> TypeName: def value(self) -> Number:
return "c" """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): class Addition(AbstractOperation):
...@@ -48,134 +44,86 @@ class Addition(AbstractOperation): ...@@ -48,134 +44,86 @@ class Addition(AbstractOperation):
TODO: More info. TODO: More info.
""" """
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""): def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(name) super().__init__(input_count=2, output_count=1,
name=name, input_sources=[src0, src1])
self._input_ports = [InputPort(0, self), InputPort(1, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None: @classmethod
self._input_ports[0].connect(source1) def type_name(cls) -> TypeName:
if source2 is not None: return "add"
self._input_ports[1].connect(source2)
def evaluate(self, a, b): def evaluate(self, a, b):
return a + b return a + b
@property
def type_name(self) -> TypeName:
return "add"
class Subtraction(AbstractOperation): class Subtraction(AbstractOperation):
"""Binary subtraction operation. """Binary subtraction operation.
TODO: More info. TODO: More info.
""" """
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""): def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(name) super().__init__(input_count=2, output_count=1,
self._input_ports = [InputPort(0, self), InputPort(1, self)] name=name, input_sources=[src0, src1])
self._output_ports = [OutputPort(0, self)]
if source1 is not None: @classmethod
self._input_ports[0].connect(source1) def type_name(cls) -> TypeName:
if source2 is not None: return "sub"
self._input_ports[1].connect(source2)
def evaluate(self, a, b): def evaluate(self, a, b):
return a - b return a - b
@property
def type_name(self) -> TypeName:
return "sub"
class Multiplication(AbstractOperation): class Multiplication(AbstractOperation):
"""Binary multiplication operation. """Binary multiplication operation.
TODO: More info. TODO: More info.
""" """
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""): def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(name) super().__init__(input_count=2, output_count=1,
self._input_ports = [InputPort(0, self), InputPort(1, self)] name=name, input_sources=[src0, src1])
self._output_ports = [OutputPort(0, self)]
if source1 is not None: @classmethod
self._input_ports[0].connect(source1) def type_name(cls) -> TypeName:
if source2 is not None: return "mul"
self._input_ports[1].connect(source2)
def evaluate(self, a, b): def evaluate(self, a, b):
return a * b return a * b
@property
def type_name(self) -> TypeName:
return "mul"
class Division(AbstractOperation): class Division(AbstractOperation):
"""Binary division operation. """Binary division operation.
TODO: More info. TODO: More info.
""" """
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""): def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(name) super().__init__(input_count=2, output_count=1,
self._input_ports = [InputPort(0, self), InputPort(1, self)] name=name, input_sources=[src0, src1])
self._output_ports = [OutputPort(0, self)]
if source1 is not None: @classmethod
self._input_ports[0].connect(source1) def type_name(cls) -> TypeName:
if source2 is not None: return "div"
self._input_ports[1].connect(source2)
def evaluate(self, a, b): def evaluate(self, a, b):
return a / b return a / b
@property
def type_name(self) -> TypeName:
return "div"
class Min(AbstractOperation):
class SquareRoot(AbstractOperation): """Binary min operation.
"""Unary square root operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return sqrt((complex)(a))
@property
def type_name(self) -> TypeName:
return "sqrt"
class ComplexConjugate(AbstractOperation):
"""Unary complex conjugate operation.
TODO: More info. TODO: More info.
""" """
def __init__(self, source1: OutputPort = None, name: Name = ""): def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(name) super().__init__(input_count=2, output_count=1,
self._input_ports = [InputPort(0, self)] name=name, input_sources=[src0, src1])
self._output_ports = [OutputPort(0, self)]
if source1 is not None: @classmethod
self._input_ports[0].connect(source1) def type_name(cls) -> TypeName:
return "min"
def evaluate(self, a):
return conjugate(a)
@property def evaluate(self, a, b):
def type_name(self) -> TypeName: assert not isinstance(a, complex) and not isinstance(b, complex), \
return "conj" ("core_operations.Min does not support complex numbers.")
return a if a < b else b
class Max(AbstractOperation): class Max(AbstractOperation):
...@@ -183,155 +131,112 @@ class Max(AbstractOperation): ...@@ -183,155 +131,112 @@ class Max(AbstractOperation):
TODO: More info. TODO: More info.
""" """
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""): def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(name) super().__init__(input_count=2, output_count=1,
self._input_ports = [InputPort(0, self), InputPort(1, self)] name=name, input_sources=[src0, src1])
self._output_ports = [OutputPort(0, self)]
if source1 is not None: @classmethod
self._input_ports[0].connect(source1) def type_name(cls) -> TypeName:
if source2 is not None: return "max"
self._input_ports[1].connect(source2)
def evaluate(self, a, b): def evaluate(self, a, b):
assert not isinstance(a, complex) and not isinstance(b, complex), \ assert not isinstance(a, complex) and not isinstance(b, complex), \
("core_operations.Max does not support complex numbers.") ("core_operations.Max does not support complex numbers.")
return a if a > b else b return a if a > b else b
@property
def type_name(self) -> TypeName:
return "max"
class Min(AbstractOperation):
"""Binary min 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: class SquareRoot(AbstractOperation):
self._input_ports[0].connect(source1) """Unary square root operation.
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
@property
def type_name(self) -> TypeName:
return "min"
class Absolute(AbstractOperation):
"""Unary absolute value operation.
TODO: More info. TODO: More info.
""" """
def __init__(self, source1: OutputPort = None, name: Name = ""): def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(name) super().__init__(input_count=1, output_count=1,
self._input_ports = [InputPort(0, self)] name=name, input_sources=[src0])
self._output_ports = [OutputPort(0, self)]
if source1 is not None: @classmethod
self._input_ports[0].connect(source1) def type_name(cls) -> TypeName:
return "sqrt"
def evaluate(self, a): def evaluate(self, a):
return np_abs(a) return sqrt(complex(a))
@property
def type_name(self) -> TypeName:
return "abs"
class ComplexConjugate(AbstractOperation):
class ConstantMultiplication(AbstractOperation): """Unary complex conjugate operation.
"""Unary constant multiplication operation.
TODO: More info. TODO: More info.
""" """
def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""): def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(name) super().__init__(input_count=1, output_count=1,
self._input_ports = [InputPort(0, self)] name=name, input_sources=[src0])
self._output_ports = [OutputPort(0, self)]
self._parameters["coefficient"] = coefficient
if source1 is not None: @classmethod
self._input_ports[0].connect(source1) def type_name(cls) -> TypeName:
return "conj"
def evaluate(self, a): def evaluate(self, a):
return a * self.param("coefficient") return conjugate(a)
@property
def type_name(self) -> TypeName:
return "cmul"
class ConstantAddition(AbstractOperation): class Absolute(AbstractOperation):
"""Unary constant addition operation. """Unary absolute value operation.
TODO: More info. TODO: More info.
""" """
def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""): def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(name) super().__init__(input_count=1, output_count=1,
self._input_ports = [InputPort(0, self)] name=name, input_sources=[src0])
self._output_ports = [OutputPort(0, self)]
self._parameters["coefficient"] = coefficient
if source1 is not None: @classmethod
self._input_ports[0].connect(source1) def type_name(cls) -> TypeName:
return "abs"
def evaluate(self, a): def evaluate(self, a):
return a + self.param("coefficient") return np_abs(a)
@property
def type_name(self) -> TypeName:
return "cadd"
class ConstantSubtraction(AbstractOperation): class ConstantMultiplication(AbstractOperation):
"""Unary constant subtraction operation. """Unary constant multiplication operation.
TODO: More info. TODO: More info.
""" """
def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""): def __init__(self, value: Number = 0, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(name) super().__init__(input_count=1, output_count=1,
self._input_ports = [InputPort(0, self)] name=name, input_sources=[src0])
self._output_ports = [OutputPort(0, self)] self.set_param("value", value)
self._parameters["coefficient"] = coefficient
if source1 is not None: @classmethod
self._input_ports[0].connect(source1) def type_name(cls) -> TypeName:
return "cmul"
def evaluate(self, a): def evaluate(self, a):
return a - self.param("coefficient") return a * self.param("value")
@property @property
def type_name(self) -> TypeName: def value(self) -> Number:
return "csub" """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 ConstantDivision(AbstractOperation): class Butterfly(AbstractOperation):
"""Unary constant division operation. """Butterfly operation that returns two outputs.
The first output is a + b and the second output is a - b.
TODO: More info. TODO: More info.
""" """
def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""): def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(name) super().__init__(input_count=2, output_count=2,
self._input_ports = [InputPort(0, self)] name=name, input_sources=[src0, src1])
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): @classmethod
return a / self.param("coefficient") def type_name(cls) -> TypeName:
return "bfly"
@property def evaluate(self, a, b):
def type_name(self) -> TypeName: return a + b, a - b
return "cdiv"
b_asic/graph_component.py
View file @ 5ca593d6
...@@ -4,10 +4,15 @@ TODO: More info. ...@@ -4,10 +4,15 @@ TODO: More info.
""" """
from abc import ABC, abstractmethod 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) Name = NewType("Name", str)
TypeName = NewType("TypeName", str) TypeName = NewType("TypeName", str)
GraphID = NewType("GraphID", str)
GraphIDNumber = NewType("GraphIDNumber", int)
class GraphComponent(ABC): class GraphComponent(ABC):
...@@ -15,35 +20,90 @@ class GraphComponent(ABC): ...@@ -15,35 +20,90 @@ class GraphComponent(ABC):
TODO: More info. TODO: More info.
""" """
@property @classmethod
@abstractmethod @abstractmethod
def type_name(self) -> TypeName: def type_name(cls) -> TypeName:
"""Return the type name of the graph component""" """Get the type name of this graph component"""
raise NotImplementedError raise NotImplementedError
@property @property
@abstractmethod @abstractmethod
def name(self) -> Name: def name(self) -> Name:
"""Return the name of the graph component.""" """Get the name of this graph component."""
raise NotImplementedError raise NotImplementedError
@name.setter @name.setter
@abstractmethod @abstractmethod
def name(self, name: Name) -> None: 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 raise NotImplementedError
class AbstractGraphComponent(GraphComponent): class AbstractGraphComponent(GraphComponent):
"""Abstract Graph Component class which is a component of a signal flow graph. """Abstract Graph Component class which is a component of a signal flow graph.
TODO: More info. TODO: More info.
""" """
_name: Name _name: Name
_graph_id: GraphID
_parameters: Dict[str, Any]
def __init__(self, name: Name = ""): def __init__(self, name: Name = ""):
self._name = name self._name = name
self._graph_id = ""
self._parameters = {}
@property @property
def name(self) -> Name: def name(self) -> Name:
...@@ -52,3 +112,41 @@ class AbstractGraphComponent(GraphComponent): ...@@ -52,3 +112,41 @@ class AbstractGraphComponent(GraphComponent):
@name.setter @name.setter
def name(self, name: Name) -> None: def name(self, name: Name) -> None:
self._name = name self._name = name
@property
def graph_id(self) -> GraphID:
return self._graph_id
@graph_id.setter
def graph_id(self, graph_id: GraphID) -> None:
self._graph_id = graph_id
@property
def params(self) -> Mapping[str, Any]:
return self._parameters.copy()
def param(self, name: str) -> Any:
return self._parameters.get(name)
def set_param(self, name: str, value: Any) -> None:
self._parameters[name] = value
def copy_component(self, *args, **kwargs) -> GraphComponent:
new_component = self.__class__(*args, **kwargs)
new_component.name = copy(self.name)
new_component.graph_id = copy(self.graph_id)
for name, value in self.params.items():
new_component.set_param(copy(name), deepcopy(value)) # pylint: disable=no-member
return new_component
def traverse(self) -> Generator[GraphComponent, None, None]:
# Breadth first search.
visited = {self}
fontier = deque([self])
while fontier:
component = fontier.popleft()
yield component
for neighbor in component.neighbors:
if neighbor not in visited:
visited.add(neighbor)
fontier.append(neighbor)
b_asic/graph_id.py 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 @ 5ca593d6
...@@ -3,93 +3,187 @@ B-ASIC Operation Module. ...@@ -3,93 +3,187 @@ B-ASIC Operation Module.
TODO: More info. TODO: More info.
""" """
import collections
from abc import abstractmethod from abc import abstractmethod
from numbers import Number from numbers import Number
from typing import List, Dict, Optional, Any, Set, TYPE_CHECKING from typing import NewType, List, Sequence, Iterable, Mapping, MutableMapping, Optional, Any, Set, Union
from collections import deque from math import trunc
from b_asic.graph_component import GraphComponent, AbstractGraphComponent, Name 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 from b_asic.signal import Signal
if TYPE_CHECKING: OutputKey = NewType("OutputKey", str)
from b_asic.port import InputPort, OutputPort OutputMap = Mapping[OutputKey, Optional[Number]]
MutableOutputMap = MutableMapping[OutputKey, Optional[Number]]
RegisterMap = Mapping[OutputKey, Number]
MutableRegisterMap = MutableMapping[OutputKey, Number]
class Operation(GraphComponent): class Operation(GraphComponent, SignalSourceProvider):
"""Operation interface. """Operation interface.
TODO: More info. TODO: More info.
""" """
@abstractmethod @abstractmethod
def inputs(self) -> "List[InputPort]": def __add__(self, src: Union[SignalSourceProvider, Number]) -> "Addition":
"""Get a list of all input ports.""" """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 raise NotImplementedError
@abstractmethod @abstractmethod
def outputs(self) -> "List[OutputPort]": def __mul__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Multiplication, ConstantMultiplication]":
"""Get a list of all output ports.""" """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 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 @abstractmethod
def input_count(self) -> int: def input_count(self) -> int:
"""Get the number of input ports.""" """Get the number of input ports."""
raise NotImplementedError raise NotImplementedError
@property
@abstractmethod @abstractmethod
def output_count(self) -> int: def output_count(self) -> int:
"""Get the number of output ports.""" """Get the number of output ports."""
raise NotImplementedError raise NotImplementedError
@abstractmethod @abstractmethod
def input(self, i: int) -> "InputPort": def input(self, index: int) -> InputPort:
"""Get the input port at index i.""" """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 inputs(self) -> Sequence[InputPort]:
"""Get all input ports."""
raise NotImplementedError
@property
@abstractmethod
def outputs(self) -> Sequence[OutputPort]:
"""Get all output ports."""
raise NotImplementedError
@property
@abstractmethod
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 output_signals(self) -> Iterable[Signal]:
"""Get all the signals that are connected to this operation's output ports,
in no particular order.
"""
raise NotImplementedError raise NotImplementedError
@abstractmethod @abstractmethod
def output(self, i: int) -> "OutputPort": def key(self, index: int, prefix: str = "") -> OutputKey:
"""Get the output port at index i.""" """Get the key used to access the output of a certain output of this operation
from the output parameter passed to current_output(s) or evaluate_output(s).
"""
raise NotImplementedError raise NotImplementedError
@abstractmethod @abstractmethod
def params(self) -> Dict[str, Optional[Any]]: def current_output(self, index: int, registers: Optional[RegisterMap] = None, prefix: str = "") -> Optional[Number]:
"""Get a dictionary of all parameter values.""" """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 raise NotImplementedError
@abstractmethod @abstractmethod
def param(self, name: str) -> Optional[Any]: def evaluate_output(self, index: int, input_values: Sequence[Number], results: Optional[MutableOutputMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = "") -> Number:
"""Get the value of a parameter. """Evaluate the output at the given index of this operation with the given input values.
Returns None if the parameter is not defined. 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 raise NotImplementedError
@abstractmethod @abstractmethod
def set_param(self, name: str, value: Any) -> None: def current_outputs(self, registers: Optional[RegisterMap] = None, prefix: str = "") -> Sequence[Optional[Number]]:
"""Set the value of a parameter. """Get all current outputs of this operation, if available.
The parameter must be defined. See current_output for more information.
""" """
raise NotImplementedError raise NotImplementedError
@abstractmethod @abstractmethod
def evaluate_outputs(self, state: "SimulationState") -> List[Number]: def evaluate_outputs(self, input_values: Sequence[Number], results: Optional[MutableOutputMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = "") -> Sequence[Number]:
"""Simulate the circuit until its iteration count matches that of the simulation state, """Evaluate all outputs of this operation given the input values.
then return the resulting output vector. See evaluate_output for more information.
""" """
raise NotImplementedError raise NotImplementedError
@abstractmethod @abstractmethod
def split(self) -> "List[Operation]": def split(self) -> Iterable["Operation"]:
"""Split the operation into multiple operations. """Split the operation into multiple operations.
If splitting is not possible, this may return a list containing only the operation itself. If splitting is not possible, this may return a list containing only the operation itself.
""" """
raise NotImplementedError raise NotImplementedError
@property
@abstractmethod @abstractmethod
def neighbors(self) -> "List[Operation]": def inputs_required_for_output(self, output_index: int) -> Iterable[int]:
"""Return all operations that are connected by signals to this operation. """Get the input indices of all inputs in this operation whose values are required in order to evalueate the output at the given output index."""
If no neighbors are found, this returns an empty list.
"""
raise NotImplementedError raise NotImplementedError
...@@ -98,171 +192,212 @@ class AbstractOperation(Operation, AbstractGraphComponent): ...@@ -98,171 +192,212 @@ class AbstractOperation(Operation, AbstractGraphComponent):
TODO: More info. TODO: More info.
""" """
_input_ports: List["InputPort"] _input_ports: List[InputPort]
_output_ports: List["OutputPort"] _output_ports: List[OutputPort]
_parameters: Dict[str, Optional[Any]]
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) super().__init__(name)
self._input_ports = []
self._output_ports = [] self._input_ports = [InputPort(self, i) for i in range(input_count)]
self._parameters = {} self._output_ports = [OutputPort(self, i) for i in range(output_count)]
# 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 @abstractmethod
def evaluate(self, *inputs) -> Any: # pylint: disable=arguments-differ def evaluate(self, *inputs) -> Any: # pylint: disable=arguments-differ
"""Evaluate the operation and generate a list of output values given a """Evaluate the operation and generate a list of output values given a list of input values."""
list of input values.
"""
raise NotImplementedError raise NotImplementedError
def inputs(self) -> List["InputPort"]: def __add__(self, src: Union[SignalSourceProvider, Number]) -> "Addition":
return self._input_ports.copy() # Import here to avoid circular imports.
from b_asic.core_operations import Constant, Addition
return Addition(self, Constant(src) if isinstance(src, Number) else src)
def __radd__(self, src: Union[SignalSourceProvider, Number]) -> "Addition":
# Import here to avoid circular imports.
from b_asic.core_operations import Constant, Addition
return Addition(Constant(src) if isinstance(src, Number) else src, self)
def __sub__(self, src: Union[SignalSourceProvider, Number]) -> "Subtraction":
# Import here to avoid circular imports.
from b_asic.core_operations import Constant, Subtraction
return Subtraction(self, Constant(src) if isinstance(src, Number) else src)
def __rsub__(self, src: Union[SignalSourceProvider, Number]) -> "Subtraction":
# Import here to avoid circular imports.
from b_asic.core_operations import Constant, Subtraction
return Subtraction(Constant(src) if isinstance(src, Number) else src, self)
def __mul__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Multiplication, ConstantMultiplication]":
# Import here to avoid circular imports.
from b_asic.core_operations import Multiplication, ConstantMultiplication
return ConstantMultiplication(src, self) if isinstance(src, Number) else Multiplication(self, src)
def outputs(self) -> List["OutputPort"]: def __rmul__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Multiplication, ConstantMultiplication]":
return self._output_ports.copy() # Import here to avoid circular imports.
from b_asic.core_operations import Multiplication, ConstantMultiplication
return ConstantMultiplication(src, self) if isinstance(src, Number) else Multiplication(src, self)
def __truediv__(self, src: Union[SignalSourceProvider, Number]) -> "Division":
# Import here to avoid circular imports.
from b_asic.core_operations import Constant, Division
return Division(self, Constant(src) if isinstance(src, Number) else src)
def __rtruediv__(self, src: Union[SignalSourceProvider, Number]) -> "Division":
# Import here to avoid circular imports.
from b_asic.core_operations import Constant, Division
return Division(Constant(src) if isinstance(src, Number) else src, self)
@property
def input_count(self) -> int: def input_count(self) -> int:
return len(self._input_ports) return len(self._input_ports)
@property
def output_count(self) -> int: def output_count(self) -> int:
return len(self._output_ports) return len(self._output_ports)
def input(self, i: int) -> "InputPort": def input(self, index: int) -> InputPort:
return self._input_ports[i] return self._input_ports[index]
def output(self, i: int) -> "OutputPort": def output(self, index: int) -> OutputPort:
return self._output_ports[i] return self._output_ports[index]
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]
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]
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
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 @property
def neighbors(self) -> List[Operation]: def inputs(self) -> Sequence[InputPort]:
neighbors: List[Operation] = [] return self._input_ports
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
if isinstance(other, Operation): @property
return Addition(self.output(0), other.output(0)) def outputs(self) -> Sequence[OutputPort]:
elif isinstance(other, Number): return self._output_ports
return ConstantAddition(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")
def __sub__(self, other): @property
"""Overloads the subtraction operator to make it return a new Subtraction operation def input_signals(self) -> Iterable[Signal]:
object that is connected to the self and other objects. If other is a number then result = []
returns a ConstantSubtraction operation object instead. for p in self.inputs:
""" for s in p.signals:
# Import here to avoid circular imports. result.append(s)
from b_asic.core_operations import Subtraction, ConstantSubtraction return result
if isinstance(other, Operation): @property
return Subtraction(self.output(0), other.output(0)) def output_signals(self) -> Iterable[Signal]:
elif isinstance(other, Number): result = []
return ConstantSubtraction(other, self.output(0)) for p in self.outputs:
for s in p.signals:
result.append(s)
return result
def key(self, index: int, prefix: str = "") -> OutputKey:
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[MutableOutputMap] = 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: else:
raise TypeError("Other type is not an Operation or a Number.") raise RuntimeError(
f"Operation evaluated to invalid type (expected Sequence/Number, got {values.__class__.__name__})")
def __mul__(self, other): if self.output_count == 1:
"""Overloads the multiplication operator to make it return a new Multiplication operation results[self.key(index, prefix)] = values[index]
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
if isinstance(other, Operation):
return Multiplication(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantMultiplication(other, self.output(0))
else: 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 __truediv__(self, other): def current_outputs(self, registers: Optional[RegisterMap] = None, prefix: str = "") -> Sequence[Optional[Number]]:
"""Overloads the division operator to make it return a new Division operation return [self.current_output(i, registers, prefix) for i in range(self.output_count)]
object that is connected to the self and other objects. If other is a number then
returns a ConstantDivision operation object instead. def evaluate_outputs(self, input_values: Sequence[Number], results: Optional[MutableOutputMap] = 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 split(self) -> Iterable[Operation]:
# Import here to avoid circular imports. # 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): def inputs_required_for_output(self, output_index: int) -> Iterable[int]:
return Division(self.output(0), other.output(0)) if output_index < 0 or output_index >= self.output_count:
elif isinstance(other, Number): raise IndexError(f"Output index out of range (expected 0-{self.output_count - 1}, got {output_index})")
return ConstantDivision(other, self.output(0)) return [i for i in range(self.input_count)] # By default, assume each output depends on all inputs.
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 @ 5ca593d6
...@@ -4,12 +4,15 @@ TODO: More info. ...@@ -4,12 +4,15 @@ TODO: More info.
""" """
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from typing import NewType, Optional, List from copy import copy
from typing import Optional, List, Iterable, TYPE_CHECKING
from b_asic.operation import Operation
from b_asic.signal import Signal from b_asic.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): class Port(ABC):
"""Port Interface. """Port Interface.
...@@ -19,59 +22,33 @@ class Port(ABC): ...@@ -19,59 +22,33 @@ class Port(ABC):
@property @property
@abstractmethod @abstractmethod
def operation(self) -> Operation: def operation(self) -> "Operation":
"""Return the connected operation.""" """Return the connected operation."""
raise NotImplementedError raise NotImplementedError
@property @property
@abstractmethod @abstractmethod
def index(self) -> PortIndex: def index(self) -> int:
"""Return the unique PortIndex.""" """Return the index of the port."""
raise NotImplementedError raise NotImplementedError
@property @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 @abstractmethod
def signal_count(self) -> int: def signal_count(self) -> int:
"""Return the number of connected signals.""" """Return the number of connected signals."""
raise NotImplementedError raise NotImplementedError
@property
@abstractmethod @abstractmethod
def connect(self, port: "Port") -> Signal: def signals(self) -> Iterable[Signal]:
"""Create and return a signal that is connected to this port and the entered """Return all connected signals."""
port and connect this port to the signal and the entered port to the signal."""
raise NotImplementedError raise NotImplementedError
@abstractmethod @abstractmethod
def add_signal(self, signal: Signal) -> None: def add_signal(self, signal: Signal) -> None:
"""Connect this port to the entered signal. If the entered signal isn't connected to """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.""" 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."""
raise NotImplementedError raise NotImplementedError
@abstractmethod @abstractmethod
...@@ -97,22 +74,34 @@ class AbstractPort(Port): ...@@ -97,22 +74,34 @@ class AbstractPort(Port):
Handles functionality for port id and saves the connection to the parent operation. Handles functionality for port id and saves the connection to the parent operation.
""" """
_operation: "Operation"
_index: int _index: int
_operation: Operation
def __init__(self, index: int, operation: Operation): def __init__(self, operation: "Operation", index: int):
self._index = index
self._operation = operation self._operation = operation
self._index = index
@property @property
def operation(self) -> Operation: def operation(self) -> "Operation":
return self._operation return self._operation
@property @property
def index(self) -> PortIndex: def index(self) -> int:
return self._index 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): class InputPort(AbstractPort):
"""Input port. """Input port.
TODO: More info. TODO: More info.
...@@ -120,104 +109,82 @@ class InputPort(AbstractPort): ...@@ -120,104 +109,82 @@ class InputPort(AbstractPort):
_source_signal: Optional[Signal] _source_signal: Optional[Signal]
def __init__(self, port_id: PortIndex, operation: Operation): def __init__(self, operation: "Operation", index: int):
super().__init__(port_id, operation) super().__init__(operation, index)
self._source_signal = None self._source_signal = None
@property @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: def signal_count(self) -> int:
return 0 if self._source_signal is None else 1 return 0 if self._source_signal is None else 1
def connect(self, port: "OutputPort") -> Signal: @property
assert self._source_signal is None, "Connecting new port to already connected input port." def signals(self) -> Iterable[Signal]:
return Signal(port, self) # self._source_signal is set by the signal constructor return [] if self._source_signal is None else [self._source_signal]
def add_signal(self, signal: Signal) -> None: def add_signal(self, signal: Signal) -> None:
assert self._source_signal is None, "Connecting new port to already connected input port." assert self._source_signal is None, "Input port may have only one signal added."
self._source_signal: Signal = signal assert signal is not self._source_signal, "Attempted to add already connected signal."
if self is not signal.destination: self._source_signal = signal
# Connect this inputport as destination for this signal if it isn't already. signal.set_destination(self)
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()
def remove_signal(self, signal: Signal) -> None: 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 self._source_signal = None
if self is old_signal.destination: signal.remove_destination()
# Disconnect the dest of the signal if this inputport currently is the dest
old_signal.remove_destination()
def clear(self) -> None: 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. """Output port.
TODO: More info. TODO: More info.
""" """
_destination_signals: List[Signal] _destination_signals: List[Signal]
def __init__(self, port_id: PortIndex, operation: Operation): def __init__(self, operation: "Operation", index: int):
super().__init__(port_id, operation) super().__init__(operation, index)
self._destination_signals = [] self._destination_signals = []
@property @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: def signal_count(self) -> int:
return len(self._destination_signals) return len(self._destination_signals)
def connect(self, port: InputPort) -> Signal: @property
return Signal(self, port) # Signal is added to self._destination_signals in signal constructor def signals(self) -> Iterable[Signal]:
return self._destination_signals
def add_signal(self, signal: Signal) -> None: def add_signal(self, signal: Signal) -> None:
assert signal not in self.signals, \ assert signal not in self._destination_signals, "Attempted to add already connected signal."
"Attempting to connect to Signal already connected."
self._destination_signals.append(signal) self._destination_signals.append(signal)
if self is not signal.source: signal.set_source(self)
# 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
def remove_signal(self, signal: Signal) -> None: def remove_signal(self, signal: Signal) -> None:
i: int = self._destination_signals.index(signal) assert signal in self._destination_signals, "Attempted to remove already removed signal."
old_signal: Signal = self._destination_signals[i] self._destination_signals.remove(signal)
del self._destination_signals[i] signal.remove_source()
if self is old_signal.source:
old_signal.remove_source()
def clear(self) -> None: def clear(self) -> None:
for signal in self._destination_signals: for signal in copy(self._destination_signals):
self.remove_signal(signal) self.remove_signal(signal)
@property
def source(self) -> "OutputPort":
return self
b_asic/precedence_chart.py deleted 100644 → 0
View file @ 8ed17a9b
"""@package docstring
B-ASIC Precedence Chart Module.
TODO: More info.
"""
from b_asic.signal_flow_graph import SFG
class PrecedenceChart:
"""Precedence chart constructed from a signal flow graph.
TODO: More info.
"""
sfg: SFG
# TODO: More members.
def __init__(self, sfg: SFG):
self.sfg = sfg
# TODO: Implement.
# TODO: More stuff.
b_asic/schema.py deleted 100644 → 0
View file @ 8ed17a9b
"""@package docstring
B-ASIC Schema Module.
TODO: More info.
"""
from b_asic.precedence_chart import PrecedenceChart
class Schema:
"""Schema constructed from a precedence chart.
TODO: More info.
"""
pc: PrecedenceChart
# TODO: More members.
def __init__(self, pc: PrecedenceChart):
self.pc = pc
# TODO: Implement.
# TODO: More stuff.
b_asic/signal.py
View file @ 5ca593d6
"""@package docstring """@package docstring
B-ASIC Signal Module. 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: if TYPE_CHECKING:
from b_asic.port import InputPort, OutputPort from b_asic.port import InputPort, OutputPort
...@@ -12,30 +12,34 @@ if TYPE_CHECKING: ...@@ -12,30 +12,34 @@ if TYPE_CHECKING:
class Signal(AbstractGraphComponent): class Signal(AbstractGraphComponent):
"""A connection between two ports.""" """A connection between two ports."""
_source: "OutputPort" _source: Optional["OutputPort"]
_destination: "InputPort" _destination: Optional["InputPort"]
def __init__(self, source: Optional["OutputPort"] = None, \
destination: Optional["InputPort"] = None, name: Name = ""):
def __init__(self, source: Optional["OutputPort"] = None, destination: Optional["InputPort"] = None, bits: Optional[int] = None, name: Name = ""):
super().__init__(name) super().__init__(name)
self._source = None
self._source = source self._destination = None
self._destination = destination
if source is not None: if source is not None:
self.set_source(source) self.set_source(source)
if destination is not None: if destination is not None:
self.set_destination(destination) self.set_destination(destination)
self.set_param("bits", bits)
@classmethod
def type_name(cls) -> TypeName:
return "s"
@property @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 the source OutputPort of the signal."""
return self._source return self._source
@property @property
def destination(self) -> "InputPort": def destination(self) -> Optional["InputPort"]:
"""Return the destination "InputPort" of the signal.""" """Return the destination "InputPort" of the signal."""
return self._destination return self._destination
...@@ -47,11 +51,11 @@ class Signal(AbstractGraphComponent): ...@@ -47,11 +51,11 @@ class Signal(AbstractGraphComponent):
Keyword arguments: Keyword arguments:
- src: OutputPort to connect as source to the signal. - src: OutputPort to connect as source to the signal.
""" """
self.remove_source() if src is not self._source:
self._source = src self.remove_source()
if self not in src.signals: self._source = src
# If the new source isn't connected to this signal then connect it. if self not in src.signals:
src.add_signal(self) src.add_signal(self)
def set_destination(self, dest: "InputPort") -> None: def set_destination(self, dest: "InputPort") -> None:
"""Disconnect the previous destination InputPort of the signal and """Disconnect the previous destination InputPort of the signal and
...@@ -61,36 +65,44 @@ class Signal(AbstractGraphComponent): ...@@ -61,36 +65,44 @@ class Signal(AbstractGraphComponent):
Keywords argments: Keywords argments:
- dest: InputPort to connect as destination to the signal. - dest: InputPort to connect as destination to the signal.
""" """
self.remove_destination() if dest is not self._destination:
self._destination = dest self.remove_destination()
if self not in dest.signals: self._destination = dest
# If the new destination isn't connected to tis signal then connect it. if self not in dest.signals:
dest.add_signal(self) dest.add_signal(self)
@property
def type_name(self) -> TypeName:
return "s"
def remove_source(self) -> None: def remove_source(self) -> None:
"""Disconnect the source OutputPort of the signal. If the source port """Disconnect the source OutputPort of the signal. If the source port
still is connected to this signal then also disconnect the source port.""" still is connected to this signal then also disconnect the source port."""
if self._source is not None: src = self._source
old_source: "OutputPort" = self._source if src is not None:
self._source = None self._source = None
if self in old_source.signals: if self in src.signals:
# If the old destination port still is connected to this signal, then disconnect it. src.remove_signal(self)
old_source.remove_signal(self)
def remove_destination(self) -> None: def remove_destination(self) -> None:
"""Disconnect the destination InputPort of the signal.""" """Disconnect the destination InputPort of the signal."""
if self._destination is not None: dest = self._destination
old_destination: "InputPort" = self._destination if dest is not None:
self._destination = None self._destination = None
if self in old_destination.signals: if self in dest.signals:
# If the old destination port still is connected to this signal, then disconnect it. dest.remove_signal(self)
old_destination.remove_signal(self)
def is_connected(self) -> bool: def dangling(self) -> bool:
"""Returns true if the signal is connected to both a source and a destination, """Returns true if the signal is missing either a source or a destination,
else false.""" else false."""
return self._source is not None and self._destination is not None return self._source is None or self._destination is None
@property
def bits(self) -> Optional[int]:
"""Get the number of bits that this operations using this signal as an input should truncate received values to.
None = unlimited."""
return self.param("bits")
@bits.setter
def bits(self, bits: Optional[int]) -> None:
"""Set the number of bits that operations using this signal as an input should truncate received values to.
None = unlimited."""
assert bits is None or (isinstance(bits, int)
and bits >= 0), "Bits must be non-negative."
self.set_param("bits", bits)
b_asic/signal_flow_graph.py
View file @ 5ca593d6
...@@ -3,14 +3,35 @@ B-ASIC Signal Flow Graph Module. ...@@ -3,14 +3,35 @@ B-ASIC Signal Flow Graph Module.
TODO: More info. TODO: More info.
""" """
from typing import List, Dict, Optional, DefaultDict from typing import List, Iterable, Sequence, Dict, Optional, DefaultDict, MutableSet
from collections import defaultdict from numbers import Number
from collections import defaultdict, deque
from b_asic.operation import Operation from b_asic.port import SignalSourceProvider, OutputPort, InputPort
from b_asic.operation import AbstractOperation from b_asic.operation import Operation, AbstractOperation, MutableOutputMap, MutableRegisterMap
from b_asic.signal import Signal from b_asic.signal import Signal
from b_asic.graph_id import GraphIDGenerator, GraphID from b_asic.graph_component import GraphID, GraphIDNumber, GraphComponent, Name, TypeName
from b_asic.graph_component import GraphComponent, Name, TypeName from b_asic.special_operations import Input, Output, Register
from b_asic.core_operations import Constant
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): class SFG(AbstractOperation):
...@@ -18,74 +39,589 @@ class SFG(AbstractOperation): ...@@ -18,74 +39,589 @@ class SFG(AbstractOperation):
TODO: More info. TODO: More info.
""" """
_graph_components_by_id: Dict[GraphID, GraphComponent] _components_by_id: Dict[GraphID, GraphComponent]
_graph_components_by_name: DefaultDict[Name, List[GraphComponent]] _components_by_name: DefaultDict[Name, List[GraphComponent]]
_components_ordered: List[GraphComponent]
_operations_ordered: List[Operation]
_graph_id_generator: GraphIDGenerator _graph_id_generator: GraphIDGenerator
_input_operations: List[Input]
_output_operations: List[Output]
_original_components_to_new: MutableSet[GraphComponent]
_original_input_signals_to_indices: Dict[Signal, int]
_original_output_signals_to_indices: Dict[Signal, int]
_precedence_list: Optional[List[List[OutputPort]]]
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._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 = {}
self._precedence_list = None
# 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
# 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
def __init__(self, input_signals: List[Signal] = None, output_signals: List[Signal] = None, \ self._input_operations.append(new_input_op)
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 = []
self._graph_components_by_id = dict() # Maps Graph ID to objects # Setup output signals.
self._graph_components_by_name = defaultdict(list) # Maps Name to objects if output_signals is not None:
self._graph_id_generator = GraphIDGenerator() 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))
for operation in ops: self._output_operations.append(new_output_op)
self._add_graph_component(operation) self._original_output_signals_to_indices[signal] = output_index
for input_signal in input_signals: # Setup output operations, starting from indices after output signals.
self._add_graph_component(input_signal) 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)
# TODO: Construct SFG based on what inputs that were given new_signal.set_destination(new_output_op.input(0))
# TODO: Traverse the graph between the inputs/outputs and add to self._operations. self._original_output_signals_to_indices[signal] = output_index
# TODO: Connect ports with signals with appropriate IDs.
def evaluate(self, *inputs) -> list: self._output_operations.append(new_output_op)
return [] # TODO: Implement
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 isinstance(component, Constant):
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
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)
@classmethod
def type_name(cls) -> TypeName:
return "sfg"
def _add_graph_component(self, graph_component: GraphComponent) -> GraphID: def evaluate(self, *args):
"""Add the entered graph component to the SFG's dictionary of graph objects and result = self.evaluate_outputs(args, {}, {}, "")
return a generated GraphID for it. 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[MutableOutputMap] = 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 connect_external_signals_to_components(self) -> bool:
""" Connects any external signals to this SFG's internal operations. This SFG becomes unconnected to the SFG
it is a component off, causing it to become invalid afterwards. Returns True if succesful, False otherwise. """
if len(self.inputs) != len(self.input_operations):
raise IndexError(f"Number of inputs does not match the number of input_operations in SFG.")
if len(self.outputs) != len(self.output_operations):
raise IndexError(f"Number of outputs does not match the number of output_operations SFG.")
if len(self.input_signals) == 0:
return False
if len(self.output_signals) == 0:
return False
# For each input_signal, connect it to the corresponding operation
for port, input_operation in zip(self.inputs, self.input_operations):
dest = input_operation.output(0).signals[0].destination
dest.clear()
port.signals[0].set_destination(dest)
# For each output_signal, connect it to the corresponding operation
for port, output_operation in zip(self.outputs, self.output_operations):
src = output_operation.input(0).signals[0].source
src.clear()
port.signals[0].set_source(src)
return True
@property
def input_operations(self) -> Sequence[Operation]:
"""Get the internal input operations in the same order as their respective input ports."""
return self._input_operations
@property
def output_operations(self) -> Sequence[Operation]:
"""Get the internal output operations in the same order as their respective output ports."""
return self._output_operations
def split(self) -> Iterable[Operation]:
return self.operations
def inputs_required_for_output(self, output_index: int) -> Iterable[int]:
if output_index < 0 or output_index >= self.output_count:
raise IndexError(
f"Output index out of range (expected 0-{self.output_count - 1}, got {output_index})")
input_indexes_required = []
sfg_input_operations_to_indexes = {
input_op: index for index, input_op in enumerate(self._input_operations)}
output_op = self._output_operations[output_index]
queue = deque([output_op])
visited = set([output_op])
while queue:
op = queue.popleft()
if isinstance(op, Input):
if op in sfg_input_operations_to_indexes:
input_indexes_required.append(
sfg_input_operations_to_indexes[op])
del sfg_input_operations_to_indexes[op]
for input_port in op.inputs:
for signal in input_port.signals:
if signal.source is not None:
new_op = signal.source.operation
if new_op not in visited:
queue.append(new_op)
visited.add(new_op)
return input_indexes_required
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 get_components_with_type_name(self, type_name: TypeName) -> List[GraphComponent]:
"""Get a list with all components in this graph with the specified type_name.
Keyword arguments: Keyword arguments:
graph_component: Graph component to add to the graph. type_name: The type_name of the desired components.
""" """
# Add to name dict i = self.id_number_offset + 1
self._graph_components_by_name[graph_component.name].append(graph_component) components = []
found_comp = self.find_by_id(type_name + str(i))
while found_comp is not None:
components.append(found_comp)
i += 1
found_comp = self.find_by_id(type_name + str(i))
# Add to ID dict return components
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 find_by_id(self, graph_id: GraphID) -> Optional[GraphComponent]: 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 """Find the graph component with the specified ID.
object with the entered ID was found then return None. Returns None if the component was not found.
Keyword arguments: Keyword arguments:
graph_id: Graph ID of the wanted object. graph_id: Graph ID of the desired component.
""" """
if graph_id in self._graph_components_by_id: return self._components_by_id.get(graph_id, None)
return self._graph_components_by_id[graph_id]
return None
def find_by_name(self, name: Name) -> List[GraphComponent]: def find_by_name(self, name: Name) -> Sequence[GraphComponent]:
"""Find all graph objects that have the entered name and return them """Find all graph components with the specified name.
in a list. If no graph object with the entered name was found then return an Returns an empty sequence if no components were found.
empty list.
Keyword arguments: Keyword arguments:
name: Name of the wanted object. name: Name of the desired component(s)
""" """
return self._graph_components_by_name[name] return self._components_by_name.get(name, [])
@property def _add_component_unconnected_copy(self, original_component: GraphComponent) -> GraphComponent:
def type_name(self) -> TypeName: assert original_component not in self._original_components_to_new, "Tried to add duplicate SFG component"
return "sfg" 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)
# 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)
# 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, _id: GraphID):
"""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
_id: The GraphID to match the component to replace.
"""
_sfg_copy = self()
_component = _sfg_copy.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 _sfg_copy()
def insert_operation(self, component: Operation, output_comp_id: GraphID):
"""Insert an operation in the SFG after a given source operation.
The source operation output count must match the input count of the operation as well as the output
Then return a new deepcopy of the sfg with the inserted component.
Arguments:
component: The new component, e.g Multiplication.
output_comp_id: The source operation GraphID to connect from.
"""
# Preserve the original SFG by creating a copy.
sfg_copy = self()
output_comp = sfg_copy.find_by_id(output_comp_id)
if output_comp is None:
return None
assert not isinstance(output_comp, Output), \
"Source operation can not be an output operation."
assert len(output_comp.output_signals) == component.input_count, \
"Source operation output count does not match input count for component."
assert len(output_comp.output_signals) == component.output_count, \
"Destination operation input count does not match output for component."
for index, signal_in in enumerate(output_comp.output_signals):
destination = signal_in.destination
signal_in.set_destination(component.input(index))
destination.connect(component.output(index))
# Recreate the newly coupled SFG so that all attributes are correct.
return sfg_copy()
def explode(self) -> Tuple[Sequence[Signal, Sequence[Signal]], Sequence[Tuple[Signal, Sequence[Signal]]]:
"""Destroy the sfg by making it unusable in the future and
return all of the intermidetry operations, the input operations and the output operations.
"""
return
def _evaluate_source(self, src: OutputPort, results: MutableOutputMap, 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
def get_precedence_list(self) -> List[List[OutputPort]]:
"""Returns a Precedence list of the SFG where each element in n:th the list consists
of elements that are executed in the n:th step. If the precedence list already has been
calculated for the current SFG then returns the cached version."""
if self._precedence_list is not None:
return self._precedence_list
# Find all operations with only outputs and no inputs.
no_input_ops = list(filter(lambda op: op.input_count == 0, self.operations))
reg_ops = self.get_components_with_type_name(Register.type_name())
# Find all first iter output ports for precedence
first_iter_ports = [op.output(i) for op in (no_input_ops + reg_ops) for i in range(op.output_count)]
self._precedence_list = self._traverse_for_precedence_list(first_iter_ports)
return self._precedence_list
def _traverse_for_precedence_list(self, first_iter_ports):
# Find dependencies of output ports and input ports.
outports_per_inport = defaultdict(list)
remaining_inports_per_outport = dict()
for op in self.operations:
op_inputs = op.inputs
for out_i, outport in enumerate(op.outputs):
dependendent_indexes = op.inputs_required_for_output(out_i)
remaining_inports_per_outport[outport] = len(dependendent_indexes)
for in_i in dependendent_indexes:
outports_per_inport[op_inputs[in_i]].append(outport)
# Traverse output ports for precedence
curr_iter_ports = first_iter_ports
precedence_list = []
while curr_iter_ports:
# Add the found ports to the current iter
precedence_list.append(curr_iter_ports)
next_iter_ports = []
for outport in curr_iter_ports:
for signal in outport.signals:
new_inport = signal.destination
# Don't traverse over Registers
if new_inport is not None and not isinstance(new_inport.operation, Register):
for new_outport in outports_per_inport[new_inport]:
remaining_inports_per_outport[new_outport] -= 1
if remaining_inports_per_outport[new_outport] == 0:
next_iter_ports.append(new_outport)
curr_iter_ports = next_iter_ports
return precedence_list
b_asic/simulation.py
View file @ 5ca593d6
...@@ -3,33 +3,116 @@ B-ASIC Simulation Module. ...@@ -3,33 +3,116 @@ B-ASIC Simulation Module.
TODO: More info. TODO: More info.
""" """
from collections import defaultdict
from numbers import Number from numbers import Number
from typing import List from typing import List, Dict, DefaultDict, Callable, Sequence, Mapping, Union, Optional
from b_asic.operation import OutputKey, OutputMap
from b_asic.signal_flow_graph import SFG
class OperationState:
"""Simulation state of an operation. InputProvider = Union[Number, Sequence[Number], Callable[[int], Number]]
class Simulation:
"""Simulation.
TODO: More info. TODO: More info.
""" """
output_values: List[Number] _sfg: SFG
iteration: int _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): def __init__(self, sfg: SFG, input_providers: Optional[Sequence[Optional[InputProvider]]] = None, save_results: bool = False):
self.output_values = [] self._sfg = sfg
self.iteration = 0 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: def set_inputs(self, input_providers: Sequence[Optional[InputProvider]]) -> None:
"""Simulation state. """Set the input functions used to get values for the inputs to the internal SFG."""
TODO: More info. 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] @property
iteration: int def results(self) -> Mapping[int, OutputMap]:
"""Get a mapping of all results, including intermediate values, calculated for each iteration up until now.
The outer mapping maps from iteration number to value mapping. The value mapping maps output port identifiers to values.
Example: {0: {"c1": 3, "c2": 4, "bfly1.0": 7, "bfly1.1": -1, "0": 7}}
"""
return self._results
def __init__(self): def clear_results(self) -> None:
self.operation_states = {} """Clear all results that were saved until now."""
self.iteration = 0 self._results.clear()
# TODO: More stuff. def clear_state(self) -> None:
"""Clear all current state of the simulation, except for the results and iteration."""
self._registers.clear()
self._current_input_values = [0 for _ in range(self._sfg.input_count)]
self._latest_output_values = [0 for _ in range(self._sfg.output_count)]
b_asic/special_operations.py 0 → 100644
View file @ 5ca593d6
"""@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, OutputKey, RegisterMap, MutableOutputMap, MutableRegisterMap
from b_asic.graph_component import Name, TypeName
from b_asic.port import SignalSourceProvider
class Input(AbstractOperation):
"""Input operation.
TODO: More info.
"""
def __init__(self, name: Name = ""):
super().__init__(input_count=0, output_count=1, name=name)
self.set_param("value", 0)
@classmethod
def type_name(cls) -> TypeName:
return "in"
def evaluate(self):
return self.param("value")
@property
def value(self) -> Number:
"""Get the current value of this input."""
return self.param("value")
@value.setter
def value(self, value: Number) -> None:
"""Set the current value of this input."""
self.set_param("value", value)
class Output(AbstractOperation):
"""Output operation.
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count=1, output_count=0,
name=name, input_sources=[src0])
@classmethod
def type_name(cls) -> TypeName:
return "out"
def evaluate(self, _):
return None
class Register(AbstractOperation):
"""Unit delay operation.
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, initial_value: Number = 0, name: Name = ""):
super().__init__(input_count=1, output_count=1,
name=name, input_sources=[src0])
self.set_param("initial_value", initial_value)
@classmethod
def type_name(cls) -> TypeName:
return "reg"
def evaluate(self, a):
return self.param("initial_value")
def current_output(self, index: int, registers: Optional[RegisterMap] = None, prefix: str = "") -> Optional[Number]:
if registers is not None:
return registers.get(self.key(index, prefix), self.param("initial_value"))
return self.param("initial_value")
def evaluate_output(self, index: int, input_values: Sequence[Number], results: Optional[MutableOutputMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = "") -> Number:
if index != 0:
raise IndexError(
f"Output index out of range (expected 0-0, got {index})")
if len(input_values) != 1:
raise ValueError(
f"Wrong number of inputs supplied to SFG for evaluation (expected 1, got {len(input_values)})")
key = self.key(index, prefix)
value = self.param("initial_value")
if registers is not None:
value = registers.get(key, value)
registers[key] = self.truncate_inputs(input_values)[0]
if results is not None:
results[key] = value
return value
small_logo.png 0 → 100644
View file @ 5ca593d6
small_logo.png

39.5 KiB

src/main.cpp
View file @ 5ca593d6
#include <pybind11/pybind11.h> #include <pybind11/pybind11.h>
namespace py = pybind11; namespace py = pybind11;
namespace asic { namespace asic {
int add(int a, int b) { int add(int a, int b) {
return a + b; return a + b;
} }
int sub(int a, int b) { int sub(int a, int b) {
return a - b; return a - b;
} }
} // namespace asic } // namespace asic
PYBIND11_MODULE(_b_asic, m) { PYBIND11_MODULE(_b_asic, m) {
m.doc() = "Better ASIC Toolbox Extension Module."; 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("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")); 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 @ 5ca593d6
from test.fixtures.signal import signal, signals from test.fixtures.signal import signal, signals
from test.fixtures.operation_tree import * from test.fixtures.operation_tree import *
from test.fixtures.port import * from test.fixtures.port import *
from test.fixtures.signal_flow_graph import *
import pytest import pytest
test/fixtures/operation_tree.py
View file @ 5ca593d6
from b_asic.core_operations import Addition, Constant
from b_asic.signal import Signal
import pytest import pytest
from b_asic import Addition, Constant, Signal, Butterfly
@pytest.fixture @pytest.fixture
def operation(): def operation():
return Constant(2) 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 @pytest.fixture
def operation_tree(): def operation_tree():
"""Return a addition operation connected with 2 constants. """Valid addition operation connected with 2 constants.
---C---+ 2---+
---A |
---C---+ v
add = 2 + 3 = 5
^
|
3---+
""" """
add_oper = Addition() return Addition(Constant(2), Constant(3))
create_operation(Constant, add_oper, 0, value=2)
create_operation(Constant, add_oper, 1, value=3)
return add_oper
@pytest.fixture @pytest.fixture
def large_operation_tree(): def large_operation_tree():
"""Return a constant operation connected with a large operation tree with 3 other constants and 3 additions. """Valid addition operation connected with a large operation tree with 2 other additions and 4 constants.
---C---+ 2---+
---A---+ |
---C---+ | v
+---A add---+
---C---+ | ^ |
---A---+ | |
---C---+ 3---+ v
add = (2 + 3) + (4 + 5) = 14
4---+ ^
| |
v |
add---+
^
|
5---+
""" """
add_oper = Addition() return Addition(Addition(Constant(2), Constant(3)), Addition(Constant(4), Constant(5)))
add_oper_2 = Addition()
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) @pytest.fixture
create_operation(Constant, add_oper_2, 1, value=5) def large_operation_tree_names():
"""Valid addition operation connected with a large operation tree with 2 other additions and 4 constants.
With names.
2---+
|
v
add---+
^ |
| |
3---+ v
add = (2 + 3) + (4 + 5) = 14
4---+ ^
| |
v |
add---+
^
|
5---+
"""
return Addition(Addition(Constant(2, name="constant2"), Constant(3, name="constant3")), Addition(Constant(4, name="constant4"), Constant(5, name="constant5")))
add_oper_3 = Addition() @pytest.fixture
add_oper_signal = Signal(add_oper.output(0), add_oper_3.output(0)) def butterfly_operation_tree():
add_oper._output_ports[0].add_signal(add_oper_signal) """Valid butterfly operations connected to eachother with 3 butterfly operations and 2 constants as inputs and 2 outputs.
add_oper_3._input_ports[0].add_signal(add_oper_signal) 2 ---+ +--- (2 + 4) ---+ +--- (6 + (-2)) ---+ +--- (4 + 8) ---> out1 = 12
| | | | | |
v ^ v ^ v ^
butterfly butterfly butterfly
^ v ^ v ^ v
| | | | | |
4 ---+ +--- (2 - 4) ---+ +--- (6 - (-2)) ---+ +--- (4 - 8) ---> out2 = -4
"""
return Butterfly(*(Butterfly(*(Butterfly(Constant(2), Constant(4), name="bfly3").outputs), name="bfly2").outputs), name="bfly1")
add_oper_2_signal = Signal(add_oper_2.output(0), add_oper_3.output(0)) @pytest.fixture
add_oper_2._output_ports[0].add_signal(add_oper_2_signal) def operation_graph_with_cycle():
add_oper_3._input_ports[1].add_signal(add_oper_2_signal) """Invalid addition operation connected with an operation graph containing a cycle.
return const_oper +-+
| |
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 @ 5ca593d6
import pytest import pytest
from b_asic.port import InputPort, OutputPort
from b_asic import InputPort, OutputPort
@pytest.fixture @pytest.fixture
def input_port(): def input_port():
return InputPort(0, None) return InputPort(None, 0)
@pytest.fixture @pytest.fixture
def output_port(): 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)]