Skip to content
Snippets Groups Projects

Compare revisions

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

Source

Select target project
No results found

Target

Select target project
  • da/B-ASIC
  • lukja239/B-ASIC
  • robal695/B-ASIC
3 results
Show changes
Commits on Source (84)
Hide whitespace changes
Inline Side-by-side
Showing
with 996 additions and 667 deletions
.gitignore
View file @ e8d99dd0
.vs/
.vscode/
build*/
bin*/
logs/
dist/
CMakeLists.txt.user*
*.autosave
*.creator
*.creator.user*
\#*\#
/.emacs.desktop
/.emacs.desktop.lock
*.elc
auto-save-list
tramp
.\#*
*~
.fuse_hudden*
.directory
.Trash-*
.nfs*
Thumbs.db
Thumbs.db:encryptable
ehthumbs.db
ehthumbs_vista.db
$RECYCLE.BIN/
*.stackdump
[Dd]esktop.ini
*.egg-info
__pycache__/
env/
.vs/
.vscode/
build*/
bin*/
logs/
dist/
CMakeLists.txt.user*
*.autosave
*.creator
*.creator.user*
\#*\#
/.emacs.desktop
/.emacs.desktop.lock
*.elc
auto-save-list
tramp
.\#*
*~
.fuse_hudden*
.directory
.Trash-*
.nfs*
Thumbs.db
Thumbs.db:encryptable
ehthumbs.db
ehthumbs_vista.db
$RECYCLE.BIN/
*.stackdump
[Dd]esktop.ini
*.egg-info
__pycache__/
env/
venv/
\ No newline at end of file
b_asic/__init__.py
View file @ e8d99dd0
......@@ -4,7 +4,6 @@ TODO: More info.
"""
from b_asic.core_operations import *
from b_asic.graph_component import *
from b_asic.graph_id import *
from b_asic.operation import *
from b_asic.precedence_chart import *
from b_asic.port import *
......@@ -12,3 +11,4 @@ from b_asic.schema import *
from b_asic.signal_flow_graph import *
from b_asic.signal import *
from b_asic.simulation import *
from b_asic.special_operations import *
b_asic/core_operations.py
View file @ e8d99dd0
......@@ -4,43 +4,39 @@ TODO: More info.
"""
from numbers import Number
from typing import Any
from typing import Optional
from numpy import conjugate, sqrt, abs as np_abs
from b_asic.port import InputPort, OutputPort
from b_asic.graph_id import GraphIDType
from b_asic.port import SignalSourceProvider, InputPort, OutputPort
from b_asic.operation import AbstractOperation
from b_asic.graph_component import Name, TypeName
class Input(AbstractOperation):
"""Input operation.
TODO: More info.
"""
# TODO: Implement all functions.
@property
def type_name(self) -> TypeName:
return "in"
class Constant(AbstractOperation):
"""Constant value operation.
TODO: More info.
"""
def __init__(self, value: Number = 0, name: Name = ""):
super().__init__(name)
super().__init__(input_count = 0, output_count = 1, name = name)
self.set_param("value", value)
self._output_ports = [OutputPort(0, self)]
self._parameters["value"] = value
@property
def type_name(self) -> TypeName:
return "c"
def evaluate(self):
return self.param("value")
@property
def type_name(self) -> TypeName:
return "c"
def value(self) -> Number:
"""TODO: docstring"""
return self.param("value")
@value.setter
def value(self, value: Number):
"""TODO: docstring"""
return self.set_param("value", value)
class Addition(AbstractOperation):
......@@ -48,290 +44,228 @@ class Addition(AbstractOperation):
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self), InputPort(1, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
def evaluate(self, a, b):
return a + b
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 2, output_count = 1, name = name, input_sources = [src0, src1])
@property
def type_name(self) -> TypeName:
return "add"
def evaluate(self, a, b):
return a + b
class Subtraction(AbstractOperation):
"""Binary subtraction operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self), InputPort(1, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
def evaluate(self, a, b):
return a - b
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 2, output_count = 1, name = name, input_sources = [src0, src1])
@property
def type_name(self) -> TypeName:
return "sub"
def evaluate(self, a, b):
return a - b
class Multiplication(AbstractOperation):
"""Binary multiplication operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self), InputPort(1, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
def evaluate(self, a, b):
return a * b
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 2, output_count = 1, name = name, input_sources = [src0, src1])
@property
def type_name(self) -> TypeName:
return "mul"
def evaluate(self, a, b):
return a * b
class Division(AbstractOperation):
"""Binary division operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self), InputPort(1, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
def evaluate(self, a, b):
return a / b
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 2, output_count = 1, name = name, input_sources = [src0, src1])
@property
def type_name(self) -> TypeName:
return "div"
def evaluate(self, a, b):
return a / b
class SquareRoot(AbstractOperation):
"""Unary square root operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return sqrt((complex)(a))
def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 1, output_count = 1, name = name, input_sources = [src0])
@property
def type_name(self) -> TypeName:
return "sqrt"
def evaluate(self, a):
return sqrt(complex(a))
class ComplexConjugate(AbstractOperation):
"""Unary complex conjugate operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return conjugate(a)
def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 1, output_count = 1, name = name, input_sources = [src0])
@property
def type_name(self) -> TypeName:
return "conj"
def evaluate(self, a):
return conjugate(a)
class Max(AbstractOperation):
"""Binary max operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self), InputPort(1, self)]
self._output_ports = [OutputPort(0, self)]
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 2, output_count = 1, name = name, input_sources = [src0, src1])
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
@property
def type_name(self) -> TypeName:
return "max"
def evaluate(self, a, b):
assert not isinstance(a, complex) and not isinstance(b, complex), \
("core_operations.Max does not support complex numbers.")
return a if a > b else b
@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)]
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 2, output_count = 1, name = name, input_sources = [src0, src1])
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
@property
def type_name(self) -> TypeName:
return "min"
def evaluate(self, a, b):
assert not isinstance(a, complex) and not isinstance(b, complex), \
("core_operations.Min does not support complex numbers.")
return a if a < b else b
@property
def type_name(self) -> TypeName:
return "min"
class Absolute(AbstractOperation):
"""Unary absolute value 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 np_abs(a)
def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 1, output_count = 1, name = name, input_sources = [src0])
@property
def type_name(self) -> TypeName:
return "abs"
def evaluate(self, a):
return np_abs(a)
class ConstantMultiplication(AbstractOperation):
"""Unary constant multiplication operation.
TODO: More info.
"""
def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self)]
self._output_ports = [OutputPort(0, self)]
self._parameters["coefficient"] = coefficient
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return a * self.param("coefficient")
def __init__(self, value: Number, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 1, output_count = 1, name = name, input_sources = [src0])
self.set_param("value", value)
@property
def type_name(self) -> TypeName:
return "cmul"
def evaluate(self, a):
return a * self.param("value")
class ConstantAddition(AbstractOperation):
"""Unary constant addition operation.
TODO: More info.
"""
def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self)]
self._output_ports = [OutputPort(0, self)]
self._parameters["coefficient"] = coefficient
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return a + self.param("coefficient")
def __init__(self, value: Number, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 1, output_count = 1, name = name, input_sources = [src0])
self.set_param("value", value)
@property
def type_name(self) -> TypeName:
return "cadd"
def evaluate(self, a):
return a + self.param("value")
class ConstantSubtraction(AbstractOperation):
"""Unary constant subtraction operation.
TODO: More info.
"""
def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self)]
self._output_ports = [OutputPort(0, self)]
self._parameters["coefficient"] = coefficient
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return a - self.param("coefficient")
def __init__(self, value: Number, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 1, output_count = 1, name = name, input_sources = [src0])
self.set_param("value", value)
@property
def type_name(self) -> TypeName:
return "csub"
def evaluate(self, a):
return a - self.param("value")
class ConstantDivision(AbstractOperation):
"""Unary constant division operation.
TODO: More info.
"""
def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self)]
self._output_ports = [OutputPort(0, self)]
self._parameters["coefficient"] = coefficient
def __init__(self, value: Number, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 1, output_count = 1, name = name, input_sources = [src0])
self.set_param("value", value)
if source1 is not None:
self._input_ports[0].connect(source1)
@property
def type_name(self) -> TypeName:
return "cdiv"
def evaluate(self, a):
return a / self.param("coefficient")
return a / self.param("value")
class Butterfly(AbstractOperation):
"""Butterfly operation that returns two outputs.
The first output is a + b and the second output is a - b.
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 2, output_count = 2, name = name, input_sources = [src0, src1])
def evaluate(self, a, b):
return a + b, a - b
@property
def type_name(self) -> TypeName:
return "cdiv"
return "bfly"
b_asic/graph_component.py
View file @ e8d99dd0
......@@ -4,6 +4,7 @@ TODO: More info.
"""
from abc import ABC, abstractmethod
from copy import copy
from typing import NewType
Name = NewType("Name", str)
......@@ -33,6 +34,11 @@ class GraphComponent(ABC):
"""Set the name of the graph component to the entered name."""
raise NotImplementedError
@abstractmethod
def copy_unconnected(self) -> "GraphComponent":
"""Get a copy of this graph component, except without any connected components."""
raise NotImplementedError
class AbstractGraphComponent(GraphComponent):
"""Abstract Graph Component class which is a component of a signal flow graph.
......@@ -52,3 +58,8 @@ class AbstractGraphComponent(GraphComponent):
@name.setter
def name(self, name: Name) -> None:
self._name = name
def copy_unconnected(self) -> GraphComponent:
new_comp = self.__class__()
new_comp.name = copy(self.name)
return new_comp
\ No newline at end of file
b_asic/graph_id.py deleted 100644 → 0
View file @ 8ed17a9b
"""@package docstring
B-ASIC Graph ID module for handling IDs of different objects in a graph.
TODO: More info
"""
from collections import defaultdict
from typing import NewType, DefaultDict
GraphID = NewType("GraphID", str)
GraphIDType = NewType("GraphIDType", str)
GraphIDNumber = NewType("GraphIDNumber", int)
class GraphIDGenerator:
"""A class that generates Graph IDs for objects."""
_next_id_number: DefaultDict[GraphIDType, GraphIDNumber]
def __init__(self):
self._next_id_number = defaultdict(lambda: 1) # Initalises every key element to 1
def get_next_id(self, graph_id_type: GraphIDType) -> GraphID:
"""Return the next graph id for a certain graph id type."""
graph_id = graph_id_type + str(self._next_id_number[graph_id_type])
self._next_id_number[graph_id_type] += 1 # Increase the current id number
return graph_id
b_asic/operation.py
View file @ e8d99dd0
......@@ -3,51 +3,86 @@ B-ASIC Operation Module.
TODO: More info.
"""
import collections
from abc import abstractmethod
from copy import deepcopy
from numbers import Number
from typing import List, Dict, Optional, Any, Set, TYPE_CHECKING
from typing import List, Sequence, Iterable, Dict, Optional, Any, Set, Generator, Union
from collections import deque
from b_asic.graph_component import GraphComponent, AbstractGraphComponent, Name
from b_asic.simulation import SimulationState, OperationState
from b_asic.signal import Signal
if TYPE_CHECKING:
from b_asic.port import InputPort, OutputPort
from b_asic.port import SignalSourceProvider, InputPort, OutputPort
class Operation(GraphComponent):
class Operation(GraphComponent, SignalSourceProvider):
"""Operation interface.
TODO: More info.
"""
@abstractmethod
def inputs(self) -> "List[InputPort]":
def __add__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Addition, ConstantAddition]":
"""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.
"""
raise NotImplementedError
@abstractmethod
def __sub__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Subtraction, ConstantSubtraction]":
"""Overloads the subtraction operator to make it return a new Subtraction operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantSubtraction operation object instead.
"""
raise NotImplementedError
@abstractmethod
def __mul__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Multiplication, ConstantMultiplication]":
"""Overloads the multiplication operator to make it return a new Multiplication operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantMultiplication operation object instead.
"""
raise NotImplementedError
@abstractmethod
def __truediv__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Division, ConstantDivision]":
"""Overloads the division operator to make it return a new Division operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantDivision operation object instead.
"""
raise NotImplementedError
@property
@abstractmethod
def inputs(self) -> List[InputPort]:
"""Get a list of all input ports."""
raise NotImplementedError
@property
@abstractmethod
def outputs(self) -> "List[OutputPort]":
def outputs(self) -> List[OutputPort]:
"""Get a list of all output ports."""
raise NotImplementedError
@property
@abstractmethod
def input_count(self) -> int:
"""Get the number of input ports."""
raise NotImplementedError
@property
@abstractmethod
def output_count(self) -> int:
"""Get the number of output ports."""
raise NotImplementedError
@abstractmethod
def input(self, i: int) -> "InputPort":
def input(self, i: int) -> InputPort:
"""Get the input port at index i."""
raise NotImplementedError
@abstractmethod
def output(self, i: int) -> "OutputPort":
def output(self, i: int) -> OutputPort:
"""Get the output port at index i."""
raise NotImplementedError
......@@ -66,19 +101,23 @@ class Operation(GraphComponent):
@abstractmethod
def set_param(self, name: str, value: Any) -> None:
"""Set the value of a parameter.
The parameter must be defined.
Adds the parameter if it is not already defined.
"""
raise NotImplementedError
@abstractmethod
def evaluate_outputs(self, state: "SimulationState") -> List[Number]:
"""Simulate the circuit until its iteration count matches that of the simulation state,
then return the resulting output vector.
def evaluate_output(self, i: int, input_values: Sequence[Number]) -> Sequence[Optional[Number]]:
"""Evaluate the output at index i of this operation with the given input values.
The returned sequence contains results corresponding to each output of this operation,
where a value of None means it was not evaluated.
The value at index i is guaranteed to have been evaluated, while the others may or may not
have been evaluated depending on what is the most efficient.
For example, Butterfly().evaluate_output(1, [5, 4]) may result in either (9, 1) or (None, 1).
"""
raise NotImplementedError
@abstractmethod
def split(self) -> "List[Operation]":
def split(self) -> Iterable["Operation"]:
"""Split the operation into multiple operations.
If splitting is not possible, this may return a list containing only the operation itself.
"""
......@@ -86,28 +125,53 @@ class Operation(GraphComponent):
@property
@abstractmethod
def neighbors(self) -> "List[Operation]":
def neighbors(self) -> Iterable["Operation"]:
"""Return all operations that are connected by signals to this operation.
If no neighbors are found, this returns an empty list.
"""
raise NotImplementedError
@abstractmethod
def traverse(self) -> Generator["Operation", None, None]:
"""Get a generator that recursively iterates through all operations that are connected by signals to this operation,
as well as the ones that they are connected to.
"""
raise NotImplementedError
class AbstractOperation(Operation, AbstractGraphComponent):
"""Generic abstract operation class which most implementations will derive from.
TODO: More info.
"""
_input_ports: List["InputPort"]
_output_ports: List["OutputPort"]
_input_ports: List[InputPort]
_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)
self._input_ports = []
self._output_ports = []
self._parameters = {}
# Allocate input ports.
for i in range(input_count):
self._input_ports.append(InputPort(self, i))
# Allocate output ports.
for i in range(output_count):
self._output_ports.append(OutputPort(self, i))
# 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"Operation expected {input_count} input sources but only got {source_count}")
for i, src in enumerate(input_sources):
if src is not None:
self._input_ports[i].connect(src.source)
@abstractmethod
def evaluate(self, *inputs) -> Any: # pylint: disable=arguments-differ
"""Evaluate the operation and generate a list of output values given a
......@@ -115,24 +179,61 @@ class AbstractOperation(Operation, AbstractGraphComponent):
"""
raise NotImplementedError
def inputs(self) -> List["InputPort"]:
def __add__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Addition, ConstantAddition]":
# Import here to avoid circular imports.
from b_asic.core_operations import Addition, ConstantAddition
if isinstance(src, Number):
return ConstantAddition(src, self)
return Addition(self, src)
def __sub__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Subtraction, ConstantSubtraction]":
# Import here to avoid circular imports.
from b_asic.core_operations import Subtraction, ConstantSubtraction
if isinstance(src, Number):
return ConstantSubtraction(src, self)
return Subtraction(self, src)
def __mul__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Multiplication, ConstantMultiplication]":
# Import here to avoid circular imports.
from b_asic.core_operations import Multiplication, ConstantMultiplication
if isinstance(src, Number):
return ConstantMultiplication(src, self)
return Multiplication(self, src)
def __truediv__(self, src: Union[SignalSourceProvider, Number]) -> "Union[Division, ConstantDivision]":
# Import here to avoid circular imports.
from b_asic.core_operations import Division, ConstantDivision
if isinstance(src, Number):
return ConstantDivision(src, self)
return Division(self, src)
@property
def inputs(self) -> List[InputPort]:
return self._input_ports.copy()
def outputs(self) -> List["OutputPort"]:
@property
def outputs(self) -> List[OutputPort]:
return self._output_ports.copy()
@property
def input_count(self) -> int:
return len(self._input_ports)
@property
def output_count(self) -> int:
return len(self._output_ports)
def input(self, i: int) -> "InputPort":
def input(self, i: int) -> InputPort:
return self._input_ports[i]
def output(self, i: int) -> "OutputPort":
def output(self, i: int) -> OutputPort:
return self._output_ports[i]
@property
def params(self) -> Dict[str, Optional[Any]]:
return self._parameters.copy()
......@@ -140,63 +241,51 @@ class AbstractOperation(Operation, AbstractGraphComponent):
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
def evaluate_output(self, i: int, input_values: Sequence[Number]) -> Sequence[Optional[Number]]:
result = self.evaluate(*input_values)
if isinstance(result, collections.Sequence):
if len(result) != self.output_count:
raise RuntimeError(
"Operation evaluated to incorrect number of outputs")
return result
if isinstance(result, Number):
if self.output_count != 1:
raise RuntimeError(
"Operation evaluated to incorrect number of outputs")
return [result]
raise RuntimeError("Operation evaluated to invalid type")
def split(self) -> Iterable[Operation]:
# Import here to avoid circular imports.
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]
@property
def neighbors(self) -> List[Operation]:
neighbors: List[Operation] = []
def neighbors(self) -> Iterable[Operation]:
neighbors = []
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}
def traverse(self) -> Generator[Operation, None, None]:
# Breadth first search.
visited = {self}
queue = deque([self])
while queue:
operation = queue.popleft()
......@@ -206,63 +295,17 @@ class AbstractOperation(Operation, AbstractGraphComponent):
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):
return Addition(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantAddition(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")
def __sub__(self, other):
"""Overloads the subtraction operator to make it return a new Subtraction operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantSubtraction operation object instead.
"""
# Import here to avoid circular imports.
from b_asic.core_operations import Subtraction, ConstantSubtraction
if isinstance(other, Operation):
return Subtraction(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantSubtraction(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")
def __mul__(self, other):
"""Overloads the multiplication operator to make it return a new Multiplication operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantMultiplication operation object instead.
"""
# Import here to avoid circular imports.
from b_asic.core_operations import Multiplication, ConstantMultiplication
if isinstance(other, Operation):
return Multiplication(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantMultiplication(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")
def __truediv__(self, other):
"""Overloads the division operator to make it return a new Division operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantDivision operation object instead.
"""
# Import here to avoid circular imports.
from b_asic.core_operations import Division, ConstantDivision
if isinstance(other, Operation):
return Division(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantDivision(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")
@property
def 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 copy_unconnected(self) -> GraphComponent:
new_comp: AbstractOperation = super().copy_unconnected()
for name, value in self.params.items():
new_comp.set_param(name, deepcopy(
value)) # pylint: disable=no-member
return new_comp
b_asic/port.py
View file @ e8d99dd0
......@@ -4,12 +4,15 @@ TODO: More info.
"""
from abc import ABC, abstractmethod
from typing import NewType, Optional, List
from copy import copy
from typing import NewType, Optional, List, Iterable, TYPE_CHECKING
from b_asic.operation import Operation
from b_asic.signal import Signal
from b_asic.graph_component import Name
if TYPE_CHECKING:
from b_asic.operation import Operation
PortIndex = NewType("PortIndex", int)
class Port(ABC):
"""Port Interface.
......@@ -19,59 +22,33 @@ class Port(ABC):
@property
@abstractmethod
def operation(self) -> Operation:
def operation(self) -> "Operation":
"""Return the connected operation."""
raise NotImplementedError
@property
@abstractmethod
def index(self) -> PortIndex:
"""Return the unique PortIndex."""
raise NotImplementedError
@property
@abstractmethod
def signals(self) -> List[Signal]:
"""Return a list of all connected signals."""
raise NotImplementedError
@abstractmethod
def signal(self, i: int = 0) -> Signal:
"""Return the connected signal at index i.
Keyword argumens:
i: integer index of the signal requsted.
"""
def index(self) -> int:
"""Return the index of the port."""
raise NotImplementedError
@property
@abstractmethod
def connected_ports(self) -> List["Port"]:
"""Return a list of all connected Ports."""
raise NotImplementedError
@abstractmethod
def signal_count(self) -> int:
"""Return the number of connected signals."""
raise NotImplementedError
@property
@abstractmethod
def connect(self, port: "Port") -> Signal:
"""Create and return a signal that is connected to this port and the entered
port and connect this port to the signal and the entered port to the signal."""
def signals(self) -> Iterable[Signal]:
"""Return all connected signals."""
raise NotImplementedError
@abstractmethod
def add_signal(self, signal: Signal) -> None:
"""Connect this port to the entered signal. If the entered signal isn't connected to
this port then connect the entered signal to the port aswell."""
raise NotImplementedError
@abstractmethod
def disconnect(self, port: "Port") -> None:
"""Disconnect the entered port from the port by removing it from the ports signal.
If the entered port is still connected to this ports signal then disconnect the entered
port from the signal aswell."""
this port then connect the entered signal to the port aswell.
"""
raise NotImplementedError
@abstractmethod
......@@ -97,127 +74,131 @@ class AbstractPort(Port):
Handles functionality for port id and saves the connection to the parent operation.
"""
_operation: "Operation"
_index: int
_operation: Operation
def __init__(self, index: int, operation: Operation):
self._index = index
def __init__(self, operation: "Operation", index: int):
self._operation = operation
self._index = index
@property
def operation(self) -> Operation:
def operation(self) -> "Operation":
return self._operation
@property
def index(self) -> PortIndex:
def index(self) -> int:
return self._index
class SignalSourceProvider(ABC):
"""Signal source provider interface.
TODO: More info.
"""
@property
@abstractmethod
def source(self) -> "OutputPort":
"""Get the main source port provided by this object."""
raise NotImplementedError
class InputPort(AbstractPort):
"""Input port.
TODO: More info.
"""
_source_signal: Optional[Signal]
_value_length: Optional[int]
def __init__(self, port_id: PortIndex, operation: Operation):
super().__init__(port_id, operation)
def __init__(self, operation: "Operation", index: int):
super().__init__(operation, index)
self._source_signal = None
self._value_length = None
@property
def signals(self) -> List[Signal]:
return [] if self._source_signal is None else [self._source_signal]
def signal(self, i: int = 0) -> Signal:
assert 0 <= i < self.signal_count(), "Signal index out of bound."
assert self._source_signal is not None, "No Signal connect to InputPort."
return self._source_signal
@property
def connected_ports(self) -> List[Port]:
return [] if self._source_signal is None or self._source_signal.source is None \
else [self._source_signal.source]
def signal_count(self) -> int:
return 0 if self._source_signal is None else 1
def connect(self, port: "OutputPort") -> Signal:
assert self._source_signal is None, "Connecting new port to already connected input port."
return Signal(port, self) # self._source_signal is set by the signal constructor
@property
def signals(self) -> Iterable[Signal]:
return [] if self._source_signal is None else [self._source_signal]
def add_signal(self, signal: Signal) -> None:
assert self._source_signal is None, "Connecting new port to already connected input port."
self._source_signal: Signal = signal
if self is not signal.destination:
# Connect this inputport as destination for this signal if it isn't already.
signal.set_destination(self)
def disconnect(self, port: "OutputPort") -> None:
assert self._source_signal.source is port, "The entered port is not connected to this port."
self._source_signal.remove_source()
assert self._source_signal is None, "Input port may have only one signal added."
assert signal is not self._source_signal, "Attempted to add already connected signal."
self._source_signal = signal
signal.set_destination(self)
def remove_signal(self, signal: Signal) -> None:
old_signal: Signal = self._source_signal
assert signal is self._source_signal, "Attempted to remove already removed signal."
self._source_signal = None
if self is old_signal.destination:
# Disconnect the dest of the signal if this inputport currently is the dest
old_signal.remove_destination()
signal.remove_destination()
def clear(self) -> None:
self.remove_signal(self._source_signal)
if self._source_signal is not None:
self.remove_signal(self._source_signal)
@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)
@property
def value_length(self) -> Optional[int]:
"""Get the number of bits that this port should truncate received values to."""
return self._value_length
@value_length.setter
def value_length(self, bits: Optional[int]) -> None:
"""Set the number of bits that this port should truncate received values to."""
assert bits is None or (isinstance(
bits, int) and bits >= 0), "Value length must be non-negative."
self._value_length = bits
class OutputPort(AbstractPort):
class OutputPort(AbstractPort, SignalSourceProvider):
"""Output port.
TODO: More info.
"""
_destination_signals: List[Signal]
def __init__(self, port_id: PortIndex, operation: Operation):
super().__init__(port_id, operation)
def __init__(self, operation: "Operation", index: int):
super().__init__(operation, index)
self._destination_signals = []
@property
def signals(self) -> List[Signal]:
return self._destination_signals.copy()
def signal(self, i: int = 0) -> Signal:
assert 0 <= i < self.signal_count(), "Signal index out of bounds."
return self._destination_signals[i]
@property
def connected_ports(self) -> List[Port]:
return [signal.destination for signal in self._destination_signals \
if signal.destination is not None]
def signal_count(self) -> int:
return len(self._destination_signals)
def connect(self, port: InputPort) -> Signal:
return Signal(self, port) # Signal is added to self._destination_signals in signal constructor
@property
def signals(self) -> Iterable[Signal]:
return self._destination_signals
def add_signal(self, signal: Signal) -> None:
assert signal not in self.signals, \
"Attempting to connect to Signal already connected."
assert signal not in self._destination_signals, "Attempted to add already connected signal."
self._destination_signals.append(signal)
if self is not signal.source:
# Connect this outputport to the signal if it isn't already
signal.set_source(self)
def disconnect(self, port: InputPort) -> None:
assert port in self.connected_ports, "Attempting to disconnect port that isn't connected."
for sig in self._destination_signals:
if sig.destination is port:
sig.remove_destination()
break
signal.set_source(self)
def remove_signal(self, signal: Signal) -> None:
i: int = self._destination_signals.index(signal)
old_signal: Signal = self._destination_signals[i]
del self._destination_signals[i]
if self is old_signal.source:
old_signal.remove_source()
assert signal in self._destination_signals, "Attempted to remove already removed signal."
self._destination_signals.remove(signal)
signal.remove_source()
def clear(self) -> None:
for signal in self._destination_signals:
for signal in copy(self._destination_signals):
self.remove_signal(signal)
@property
def source(self) -> "OutputPort":
return self
b_asic/signal.py
View file @ e8d99dd0
......@@ -12,30 +12,26 @@ if TYPE_CHECKING:
class Signal(AbstractGraphComponent):
"""A connection between two ports."""
_source: "OutputPort"
_destination: "InputPort"
_source: Optional["OutputPort"]
_destination: Optional["InputPort"]
def __init__(self, source: Optional["OutputPort"] = None, \
destination: Optional["InputPort"] = None, name: Name = ""):
destination: Optional["InputPort"] = None, name: Name = ""):
super().__init__(name)
self._source = source
self._destination = destination
self._source = None
self._destination = None
if source is not None:
self.set_source(source)
if destination is not None:
self.set_destination(destination)
@property
def source(self) -> "OutputPort":
def source(self) -> Optional["OutputPort"]:
"""Return the source OutputPort of the signal."""
return self._source
@property
def destination(self) -> "InputPort":
def destination(self) -> Optional["InputPort"]:
"""Return the destination "InputPort" of the signal."""
return self._destination
......@@ -47,11 +43,11 @@ class Signal(AbstractGraphComponent):
Keyword arguments:
- src: OutputPort to connect as source to the signal.
"""
self.remove_source()
self._source = src
if self not in src.signals:
# If the new source isn't connected to this signal then connect it.
src.add_signal(self)
if src is not self._source:
self.remove_source()
self._source = src
if self not in src.signals:
src.add_signal(self)
def set_destination(self, dest: "InputPort") -> None:
"""Disconnect the previous destination InputPort of the signal and
......@@ -61,11 +57,11 @@ class Signal(AbstractGraphComponent):
Keywords argments:
- dest: InputPort to connect as destination to the signal.
"""
self.remove_destination()
self._destination = dest
if self not in dest.signals:
# If the new destination isn't connected to tis signal then connect it.
dest.add_signal(self)
if dest is not self._destination:
self.remove_destination()
self._destination = dest
if self not in dest.signals:
dest.add_signal(self)
@property
def type_name(self) -> TypeName:
......@@ -74,23 +70,21 @@ class Signal(AbstractGraphComponent):
def remove_source(self) -> None:
"""Disconnect the source OutputPort of the signal. If the source port
still is connected to this signal then also disconnect the source port."""
if self._source is not None:
old_source: "OutputPort" = self._source
src = self._source
if src is not None:
self._source = None
if self in old_source.signals:
# If the old destination port still is connected to this signal, then disconnect it.
old_source.remove_signal(self)
if self in src.signals:
src.remove_signal(self)
def remove_destination(self) -> None:
"""Disconnect the destination InputPort of the signal."""
if self._destination is not None:
old_destination: "InputPort" = self._destination
dest = self._destination
if dest is not None:
self._destination = None
if self in old_destination.signals:
# If the old destination port still is connected to this signal, then disconnect it.
old_destination.remove_signal(self)
if self in dest.signals:
dest.remove_signal(self)
def is_connected(self) -> bool:
"""Returns true if the signal is connected to both a source and a destination,
def dangling(self) -> bool:
"""Returns true if the signal is missing either a source or a destination,
else false."""
return self._source is not None and self._destination is not None
return self._source is None or self._destination is None
b_asic/signal_flow_graph.py
View file @ e8d99dd0
......@@ -3,14 +3,33 @@ B-ASIC Signal Flow Graph Module.
TODO: More info.
"""
from typing import List, Dict, Optional, DefaultDict
from collections import defaultdict
from typing import NewType, List, Iterable, Sequence, Dict, Optional, DefaultDict, Set
from numbers import Number
from collections import defaultdict, deque
from b_asic.operation import Operation
from b_asic.operation import AbstractOperation
from b_asic.port import SignalSourceProvider, OutputPort
from b_asic.operation import Operation, AbstractOperation
from b_asic.signal import Signal
from b_asic.graph_id import GraphIDGenerator, GraphID
from b_asic.graph_component import GraphComponent, Name, TypeName
from b_asic.special_operations import Input, Output
GraphID = NewType("GraphID", str)
GraphIDNumber = NewType("GraphIDNumber", int)
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:
"""Return 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])
class SFG(AbstractOperation):
......@@ -18,51 +37,162 @@ class SFG(AbstractOperation):
TODO: More info.
"""
_graph_components_by_id: Dict[GraphID, GraphComponent]
_graph_components_by_name: DefaultDict[Name, List[GraphComponent]]
_components_by_id: Dict[GraphID, GraphComponent]
_components_by_name: DefaultDict[Name, List[GraphComponent]]
_graph_id_generator: GraphIDGenerator
_input_operations: List[Input]
_output_operations: List[Output]
_original_components_added: Set[GraphComponent]
_original_input_signals: Dict[Signal, int]
_original_output_signals: Dict[Signal, int]
def __init__(self, input_signals: List[Signal] = None, output_signals: List[Signal] = None, \
ops: List[Operation] = None, **kwds):
super().__init__(**kwds)
if input_signals is None:
input_signals = []
if output_signals is None:
output_signals = []
if ops is None:
ops = []
def __init__(self, input_signals: Sequence[Signal] = [], output_signals: Sequence[Signal] = [],
inputs: Sequence[Input] = [], outputs: Sequence[Output] = [],
id_number_offset: GraphIDNumber = 0, name: Name = "",
input_sources: Optional[Sequence[Optional[SignalSourceProvider]]] = None):
super().__init__(
input_count=len(input_signals) + len(inputs),
output_count=len(output_signals) + len(outputs),
name=name,
input_sources=input_sources)
self._graph_components_by_id = dict() # Maps Graph ID to objects
self._graph_components_by_name = defaultdict(list) # Maps Name to objects
self._graph_id_generator = GraphIDGenerator()
self._components_by_id = dict()
self._components_by_name = defaultdict(list)
self._components_in_dfs_order = []
self._graph_id_generator = GraphIDGenerator(id_number_offset)
self._input_operations = []
self._output_operations = []
# Maps original components to new copied components
self._added_components_mapping = {}
self._original_input_signals_indexes = {}
self._original_output_signals_indexes = {}
self._id_number_offset = id_number_offset
for operation in ops:
self._add_graph_component(operation)
# Setup input signals.
for input_index, sig in enumerate(input_signals):
assert sig not in self._added_components_mapping, "Duplicate input signals sent to SFG construcctor."
for input_signal in input_signals:
self._add_graph_component(input_signal)
new_input_op = self._add_component_copy_unconnected(Input())
new_sig = self._add_component_copy_unconnected(sig)
new_sig.set_source(new_input_op.output(0))
# TODO: Construct SFG based on what inputs that were given
# TODO: Traverse the graph between the inputs/outputs and add to self._operations.
# TODO: Connect ports with signals with appropriate IDs.
self._input_operations.append(new_input_op)
self._original_input_signals_indexes[sig] = input_index
def evaluate(self, *inputs) -> list:
return [] # TODO: Implement
# Setup input operations, starting from indexes ater input signals.
for input_index, input_op in enumerate(inputs, len(input_signals)):
assert input_op not in self._added_components_mapping, "Duplicate input operations sent to SFG constructor."
new_input_op = self._add_component_copy_unconnected(input_op)
def _add_graph_component(self, graph_component: GraphComponent) -> GraphID:
"""Add the entered graph component to the SFG's dictionary of graph objects and
return a generated GraphID for it.
for sig in input_op.output(0).signals:
assert sig not in self._added_components_mapping, "Duplicate input signals connected to input ports sent to SFG construcctor."
new_sig = self._add_component_copy_unconnected(sig)
new_sig.set_source(new_input_op.output(0))
Keyword arguments:
graph_component: Graph component to add to the graph.
"""
# Add to name dict
self._graph_components_by_name[graph_component.name].append(graph_component)
self._original_input_signals_indexes[sig] = input_index
self._input_operations.append(new_input_op)
# Setup output signals.
for output_ind, sig in enumerate(output_signals):
new_out = self._add_component_copy_unconnected(Output())
if sig in self._added_components_mapping:
# Signal already added when setting up inputs
new_sig = self._added_components_mapping[sig]
new_sig.set_destination(new_out.input(0))
else:
# New signal has to be created
new_sig = self._add_component_copy_unconnected(sig)
new_sig.set_destination(new_out.input(0))
self._output_operations.append(new_out)
self._original_output_signals_indexes[sig] = output_ind
# Setup output operations, starting from indexes after output signals.
for output_ind, output_op in enumerate(outputs, len(output_signals)):
assert output_op not in self._added_components_mapping, "Duplicate output operations sent to SFG constructor."
new_out = self._add_component_copy_unconnected(output_op)
for sig in output_op.input(0).signals:
if sig in self._added_components_mapping:
# Signal already added when setting up inputs
new_sig = self._added_components_mapping[sig]
new_sig.set_destination(new_out.input(0))
else:
# New signal has to be created
new_sig = self._add_component_copy_unconnected(sig)
new_sig.set_destination(new_out.input(0))
self._original_output_signals_indexes[sig] = output_ind
self._output_operations.append(new_out)
output_operations_set = set(self._output_operations)
# Add to ID dict
graph_id: GraphID = self._graph_id_generator.get_next_id(graph_component.type_name)
self._graph_components_by_id[graph_id] = graph_component
return graph_id
# Search the graph inwards from each input signal.
for sig, input_index in self._original_input_signals_indexes.items():
# Check if already added destination.
new_sig = self._added_components_mapping[sig]
if new_sig.destination is None:
if sig.destination is None:
raise ValueError(
f"Input signal #{input_index} is missing destination in SFG")
elif sig.destination.operation not in self._added_components_mapping:
self._copy_structure_from_operation_dfs(
sig.destination.operation)
else:
if new_sig.destination.operation in output_operations_set:
# Add directly connected input to output to dfs order list
self._components_in_dfs_order.extend([
new_sig.source.operation, new_sig, new_sig.destination.operation])
# Search the graph inwards from each output signal.
for sig, output_index in self._original_output_signals_indexes.items():
# Check if already added source.
new_sig = self._added_components_mapping[sig]
if new_sig.source is None:
if sig.source is None:
raise ValueError(
f"Output signal #{output_index} is missing source in SFG")
if sig.source.operation not in self._added_components_mapping:
self._copy_structure_from_operation_dfs(
sig.source.operation)
def __call__(self):
return self.deep_copy()
@property
def type_name(self) -> TypeName:
return "sfg"
def evaluate(self, *args):
if len(args) != self.input_count:
raise ValueError(
"Wrong number of inputs supplied to SFG for evaluation")
for arg, op in zip(args, self._input_operations):
op.value = arg
result = []
for op in self._output_operations:
result.append(self._evaluate_source(op.input(0).signals[0].source))
n = len(result)
return None if n == 0 else result[0] if n == 1 else result
def evaluate_output(self, i: int, input_values: Sequence[Number]) -> Sequence[Optional[Number]]:
assert i >= 0 and i < self.output_count, "Output index out of range"
result = [None] * self.output_count
result[i] = self._evaluate_source(
self._output_operations[i].input(0).signals[0].source)
return result
def split(self) -> Iterable[Operation]:
return filter(lambda comp: isinstance(comp, Operation), self._components_by_id.values())
@property
def components(self) -> Iterable[GraphComponent]:
"""Get all components of this graph in the dfs-traversal order."""
return self._components_in_dfs_order
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
......@@ -71,10 +201,7 @@ class SFG(AbstractOperation):
Keyword arguments:
graph_id: Graph ID of the wanted object.
"""
if graph_id in self._graph_components_by_id:
return self._graph_components_by_id[graph_id]
return None
return self._components_by_id.get(graph_id, None)
def find_by_name(self, name: Name) -> List[GraphComponent]:
"""Find all graph objects that have the entered name and return them
......@@ -84,8 +211,146 @@ class SFG(AbstractOperation):
Keyword arguments:
name: Name of the wanted object.
"""
return self._graph_components_by_name[name]
return self._components_by_name.get(name, [])
@property
def type_name(self) -> TypeName:
return "sfg"
def deep_copy(self) -> "SFG":
"""Returns a deep copy of self."""
copy = SFG(inputs=self._input_operations, outputs=self._output_operations,
id_number_offset=self._id_number_offset, name=super().name)
return copy
def _add_component_copy_unconnected(self, original_comp: GraphComponent) -> GraphComponent:
assert original_comp not in self._added_components_mapping, "Tried to add duplicate SFG component"
new_comp = original_comp.copy_unconnected()
self._added_components_mapping[original_comp] = new_comp
self._components_by_id[self._graph_id_generator.next_id(
new_comp.type_name)] = new_comp
self._components_by_name[new_comp.name].append(new_comp)
return new_comp
def _copy_structure_from_operation_dfs(self, start_op: Operation):
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._added_components_mapping:
new_op = self._add_component_copy_unconnected(original_op)
self._components_in_dfs_order.append(new_op)
else:
new_op = self._added_components_mapping[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_indexes:
# New signal already created during first step of constructor
new_signal = self._added_components_mapping[
original_signal]
new_signal.set_destination(
new_op.input(original_input_port.index))
self._components_in_dfs_order.extend(
[new_signal, new_signal.source.operation])
# Check if the signal has not been added before
elif original_signal not in self._added_components_mapping:
if original_signal.source is None:
raise ValueError(
"Dangling signal without source in SFG")
new_signal = self._add_component_copy_unconnected(
original_signal)
new_signal.set_destination(
new_op.input(original_input_port.index))
self._components_in_dfs_order.append(new_signal)
original_connected_op = original_signal.source.operation
# Check if connected Operation has been added before
if original_connected_op in self._added_components_mapping:
# Set source to the already added operations port
new_signal.set_source(
self._added_components_mapping[original_connected_op].output(
original_signal.source.index))
else:
# Create new operation, set signal source to it
new_connected_op = self._add_component_copy_unconnected(
original_connected_op)
new_signal.set_source(new_connected_op.output(
original_signal.source.index))
self._components_in_dfs_order.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_indexes:
# New signal already created during first step of constructor.
new_signal = self._added_components_mapping[
original_signal]
new_signal.set_source(
new_op.output(original_output_port.index))
self._components_in_dfs_order.extend(
[new_signal, new_signal.destination.operation])
# Check if signal has not been added before.
elif original_signal not in self._added_components_mapping:
if original_signal.source is None:
raise ValueError(
"Dangling signal without source in SFG")
new_signal = self._add_component_copy_unconnected(
original_signal)
new_signal.set_source(
new_op.output(original_output_port.index))
self._components_in_dfs_order.append(new_signal)
original_connected_op = original_signal.destination.operation
# Check if connected operation has been added.
if original_connected_op in self._added_components_mapping:
# Set destination to the already connected operations port
new_signal.set_destination(
self._added_components_mapping[original_connected_op].input(
original_signal.destination.index))
else:
# Create new operation, set destination to it.
new_connected_op = self._add_component_copy_unconnected(
original_connected_op)
new_signal.set_destination(new_connected_op.input(
original_signal.destination.index))
self._components_in_dfs_order.append(
new_connected_op)
# Add connected operation to the queue of operations to visist
op_stack.append(original_connected_op)
def _evaluate_source(self, src: OutputPort) -> Number:
input_values = []
for input_port in src.operation.inputs:
input_src = input_port.signals[0].source
input_values.append(self._evaluate_source(input_src))
return src.operation.evaluate_output(src.index, input_values)
b_asic/simulation.py
View file @ e8d99dd0
......@@ -4,7 +4,7 @@ TODO: More info.
"""
from numbers import Number
from typing import List
from typing import List, Dict
class OperationState:
......@@ -25,11 +25,19 @@ class SimulationState:
TODO: More info.
"""
# operation_states: Dict[OperationId, OperationState]
operation_states: Dict[int, OperationState]
iteration: int
def __init__(self):
self.operation_states = {}
op_state = OperationState()
self.operation_states = {1: op_state}
self.iteration = 0
# TODO: More stuff.
# @property
# #def iteration(self):
# return self.iteration
# @iteration.setter
# def iteration(self, new_iteration: int):
# self.iteration = new_iteration
#
# TODO: More stuff
b_asic/special_operations.py 0 → 100644
View file @ e8d99dd0
"""@package docstring
B-ASIC Special Operations Module.
TODO: More info.
"""
from numbers import Number
from typing import Optional
from b_asic.operation import AbstractOperation
from b_asic.graph_component import Name, TypeName
from b_asic.port import SignalSourceProvider
class Input(AbstractOperation):
"""Input operation.
TODO: More info.
"""
def __init__(self, name: Name = ""):
super().__init__(input_count = 0, output_count = 1, name = name)
self.set_param("value", 0)
@property
def type_name(self) -> TypeName:
return "in"
def evaluate(self):
return self.param("value")
@property
def value(self) -> Number:
"""TODO: docstring"""
return self.param("value")
@value.setter
def value(self, value: Number):
"""TODO: docstring"""
self.set_param("value", value)
class Output(AbstractOperation):
"""Output operation.
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 1, output_count = 0, name = name, input_sources=[src0])
@property
def type_name(self) -> TypeName:
return "out"
def evaluate(self):
return None
\ No newline at end of file
src/main.cpp
View file @ e8d99dd0
#include <pybind11/pybind11.h>
namespace py = pybind11;
namespace asic {
int add(int a, int b) {
return a + b;
}
int sub(int a, int b) {
return a - b;
}
} // namespace asic
PYBIND11_MODULE(_b_asic, m) {
m.doc() = "Better ASIC Toolbox Extension Module.";
m.def("add", &asic::add, "A function which adds two numbers.", py::arg("a"), py::arg("b"));
m.def("sub", &asic::sub, "A function which subtracts two numbers.", py::arg("a"), py::arg("b"));
#include <pybind11/pybind11.h>
namespace py = pybind11;
namespace asic {
int add(int a, int b) {
return a + b;
}
int sub(int a, int b) {
return a - b;
}
} // namespace asic
PYBIND11_MODULE(_b_asic, m) {
m.doc() = "Better ASIC Toolbox Extension Module.";
m.def("add", &asic::add, "A function which adds two numbers.", py::arg("a"), py::arg("b"));
m.def("sub", &asic::sub, "A function which subtracts two numbers.", py::arg("a"), py::arg("b"));
}
\ No newline at end of file
test/fixtures/operation_tree.py
View file @ e8d99dd0
......@@ -7,52 +7,24 @@ import pytest
def operation():
return Constant(2)
def create_operation(_type, dest_oper, index, **kwargs):
oper = _type(**kwargs)
oper_signal = Signal()
oper._output_ports[0].add_signal(oper_signal)
dest_oper._input_ports[index].add_signal(oper_signal)
return oper
@pytest.fixture
def operation_tree():
"""Return a addition operation connected with 2 constants.
---C---+
---A
+--A
---C---+
"""
add_oper = Addition()
create_operation(Constant, add_oper, 0, value=2)
create_operation(Constant, add_oper, 1, value=3)
return add_oper
return Addition(Constant(2), Constant(3))
@pytest.fixture
def large_operation_tree():
"""Return a constant operation connected with a large operation tree with 3 other constants and 3 additions.
"""Return an addition operation connected with a large operation tree with 2 other additions and 4 constants.
---C---+
---A---+
+--A---+
---C---+ |
+---A
---C---+ |
---A---+
+--A---+
---C---+
"""
add_oper = Addition()
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)
create_operation(Constant, add_oper_2, 1, value=5)
add_oper_3 = Addition()
add_oper_signal = Signal(add_oper.output(0), add_oper_3.output(0))
add_oper._output_ports[0].add_signal(add_oper_signal)
add_oper_3._input_ports[0].add_signal(add_oper_signal)
add_oper_2_signal = Signal(add_oper_2.output(0), add_oper_3.output(0))
add_oper_2._output_ports[0].add_signal(add_oper_2_signal)
add_oper_3._input_ports[1].add_signal(add_oper_2_signal)
return const_oper
return Addition(Addition(Constant(2), Constant(3)), Addition(Constant(4), Constant(5)))
test/fixtures/port.py
View file @ e8d99dd0
......@@ -3,8 +3,8 @@ from b_asic.port import InputPort, OutputPort
@pytest.fixture
def input_port():
return InputPort(0, None)
return InputPort(None, 0)
@pytest.fixture
def output_port():
return OutputPort(0, None)
return OutputPort(None, 0)
test/fixtures/signal.py
View file @ e8d99dd0
......@@ -9,4 +9,4 @@ def signal():
@pytest.fixture
def signals():
"""Return 3 signals with no connections."""
return [Signal() for _ in range(0,3)]
return [Signal() for _ in range(0, 3)]
test/test_core_operations.py
View file @ e8d99dd0
......@@ -2,226 +2,313 @@
B-ASIC test suite for the core operations.
"""
from b_asic.core_operations import Constant, Addition, Subtraction, Multiplication, Division, SquareRoot, ComplexConjugate, Max, Min, Absolute, ConstantMultiplication, ConstantAddition, ConstantSubtraction, ConstantDivision
from b_asic.core_operations import Constant, Addition, Subtraction, \
Multiplication, Division, SquareRoot, ComplexConjugate, Max, Min, \
Absolute, ConstantMultiplication, ConstantAddition, ConstantSubtraction, \
ConstantDivision, Butterfly
# Constant tests.
def test_constant():
constant_operation = Constant(3)
assert constant_operation.evaluate() == 3
def test_constant_negative():
constant_operation = Constant(-3)
assert constant_operation.evaluate() == -3
def test_constant_complex():
constant_operation = Constant(3+4j)
assert constant_operation.evaluate() == 3+4j
# Addition tests.
def test_addition():
test_operation = Addition()
constant_operation = Constant(3)
constant_operation_2 = Constant(5)
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == 8
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == 8
def test_addition_negative():
test_operation = Addition()
constant_operation = Constant(-3)
constant_operation_2 = Constant(-5)
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == -8
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == -8
def test_addition_complex():
test_operation = Addition()
constant_operation = Constant((3+5j))
constant_operation_2 = Constant((4+6j))
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == (7+11j)
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == (7+11j)
# Subtraction tests.
def test_subtraction():
test_operation = Subtraction()
constant_operation = Constant(5)
constant_operation_2 = Constant(3)
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == 2
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == 2
def test_subtraction_negative():
test_operation = Subtraction()
constant_operation = Constant(-5)
constant_operation_2 = Constant(-3)
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == -2
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == -2
def test_subtraction_complex():
test_operation = Subtraction()
constant_operation = Constant((3+5j))
constant_operation_2 = Constant((4+6j))
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == (-1-1j)
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == (-1-1j)
# Multiplication tests.
def test_multiplication():
test_operation = Multiplication()
constant_operation = Constant(5)
constant_operation_2 = Constant(3)
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == 15
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == 15
def test_multiplication_negative():
test_operation = Multiplication()
constant_operation = Constant(-5)
constant_operation_2 = Constant(-3)
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == 15
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == 15
def test_multiplication_complex():
test_operation = Multiplication()
constant_operation = Constant((3+5j))
constant_operation_2 = Constant((4+6j))
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == (-18+38j)
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == (-18+38j)
# Division tests.
def test_division():
test_operation = Division()
constant_operation = Constant(30)
constant_operation_2 = Constant(5)
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == 6
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == 6
def test_division_negative():
test_operation = Division()
constant_operation = Constant(-30)
constant_operation_2 = Constant(-5)
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == 6
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == 6
def test_division_complex():
test_operation = Division()
constant_operation = Constant((60+40j))
constant_operation_2 = Constant((10+20j))
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == (2.8-1.6j)
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == (2.8-1.6j)
# SquareRoot tests.
def test_squareroot():
test_operation = SquareRoot()
constant_operation = Constant(36)
assert test_operation.evaluate(constant_operation.evaluate()) == 6
def test_squareroot_negative():
test_operation = SquareRoot()
constant_operation = Constant(-36)
assert test_operation.evaluate(constant_operation.evaluate()) == 6j
def test_squareroot_complex():
test_operation = SquareRoot()
constant_operation = Constant((48+64j))
assert test_operation.evaluate(constant_operation.evaluate()) == (8+4j)
# ComplexConjugate tests.
def test_complexconjugate():
test_operation = ComplexConjugate()
constant_operation = Constant(3+4j)
assert test_operation.evaluate(constant_operation.evaluate()) == (3-4j)
def test_test_complexconjugate_negative():
test_operation = ComplexConjugate()
constant_operation = Constant(-3-4j)
assert test_operation.evaluate(constant_operation.evaluate()) == (-3+4j)
# Max tests.
def test_max():
test_operation = Max()
constant_operation = Constant(30)
constant_operation_2 = Constant(5)
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == 30
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == 30
def test_max_negative():
test_operation = Max()
constant_operation = Constant(-30)
constant_operation_2 = Constant(-5)
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == -5
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == -5
# Min tests.
def test_min():
test_operation = Min()
constant_operation = Constant(30)
constant_operation_2 = Constant(5)
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == 5
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == 5
def test_min_negative():
test_operation = Min()
constant_operation = Constant(-30)
constant_operation_2 = Constant(-5)
assert test_operation.evaluate(constant_operation.evaluate(), constant_operation_2.evaluate()) == -30
assert test_operation.evaluate(
constant_operation.evaluate(), constant_operation_2.evaluate()) == -30
# Absolute tests.
def test_absolute():
test_operation = Absolute()
constant_operation = Constant(30)
assert test_operation.evaluate(constant_operation.evaluate()) == 30
def test_absolute_negative():
test_operation = Absolute()
constant_operation = Constant(-5)
assert test_operation.evaluate(constant_operation.evaluate()) == 5
def test_absolute_complex():
test_operation = Absolute()
constant_operation = Constant((3+4j))
assert test_operation.evaluate(constant_operation.evaluate()) == 5.0
# ConstantMultiplication tests.
def test_constantmultiplication():
test_operation = ConstantMultiplication(5)
constant_operation = Constant(20)
assert test_operation.evaluate(constant_operation.evaluate()) == 100
def test_constantmultiplication_negative():
test_operation = ConstantMultiplication(5)
constant_operation = Constant(-5)
assert test_operation.evaluate(constant_operation.evaluate()) == -25
def test_constantmultiplication_complex():
test_operation = ConstantMultiplication(3+2j)
constant_operation = Constant((3+4j))
assert test_operation.evaluate(constant_operation.evaluate()) == (1+18j)
# ConstantAddition tests.
def test_constantaddition():
test_operation = ConstantAddition(5)
constant_operation = Constant(20)
assert test_operation.evaluate(constant_operation.evaluate()) == 25
def test_constantaddition_negative():
test_operation = ConstantAddition(4)
constant_operation = Constant(-5)
assert test_operation.evaluate(constant_operation.evaluate()) == -1
def test_constantaddition_complex():
test_operation = ConstantAddition(3+2j)
constant_operation = Constant((3+4j))
assert test_operation.evaluate(constant_operation.evaluate()) == (6+6j)
# ConstantSubtraction tests.
def test_constantsubtraction():
test_operation = ConstantSubtraction(5)
constant_operation = Constant(20)
assert test_operation.evaluate(constant_operation.evaluate()) == 15
def test_constantsubtraction_negative():
test_operation = ConstantSubtraction(4)
constant_operation = Constant(-5)
assert test_operation.evaluate(constant_operation.evaluate()) == -9
def test_constantsubtraction_complex():
test_operation = ConstantSubtraction(4+6j)
constant_operation = Constant((3+4j))
assert test_operation.evaluate(constant_operation.evaluate()) == (-1-2j)
# ConstantDivision tests.
def test_constantdivision():
test_operation = ConstantDivision(5)
constant_operation = Constant(20)
assert test_operation.evaluate(constant_operation.evaluate()) == 4
def test_constantdivision_negative():
test_operation = ConstantDivision(4)
constant_operation = Constant(-20)
assert test_operation.evaluate(constant_operation.evaluate()) == -5
def test_constantdivision_complex():
test_operation = ConstantDivision(2+2j)
constant_operation = Constant((10+10j))
assert test_operation.evaluate(constant_operation.evaluate()) == (5+0j)
def test_butterfly():
test_operation = Butterfly()
assert list(test_operation.evaluate(2, 3)) == [5, -1]
def test_butterfly_negative():
test_operation = Butterfly()
assert list(test_operation.evaluate(-2, -3)) == [-5, 1]
def test_buttefly_complex():
test_operation = Butterfly()
assert list(test_operation.evaluate(2+1j, 3-2j)) == [5-1j, -1+3j]
test/test_graph_id_generator.py
View file @ e8d99dd0
......@@ -2,7 +2,7 @@
B-ASIC test suite for graph id generator.
"""
from b_asic.graph_id import GraphIDGenerator, GraphID
from b_asic.signal_flow_graph import GraphIDGenerator, GraphID
import pytest
@pytest.fixture
......@@ -12,17 +12,17 @@ def graph_id_generator():
class TestGetNextId:
def test_empty_string_generator(self, graph_id_generator):
"""Test the graph id generator for an empty string type."""
assert graph_id_generator.get_next_id("") == "1"
assert graph_id_generator.get_next_id("") == "2"
assert graph_id_generator.next_id("") == "1"
assert graph_id_generator.next_id("") == "2"
def test_normal_string_generator(self, graph_id_generator):
""""Test the graph id generator for a normal string type."""
assert graph_id_generator.get_next_id("add") == "add1"
assert graph_id_generator.get_next_id("add") == "add2"
assert graph_id_generator.next_id("add") == "add1"
assert graph_id_generator.next_id("add") == "add2"
def test_different_strings_generator(self, graph_id_generator):
"""Test the graph id generator for different strings."""
assert graph_id_generator.get_next_id("sub") == "sub1"
assert graph_id_generator.get_next_id("mul") == "mul1"
assert graph_id_generator.get_next_id("sub") == "sub2"
assert graph_id_generator.get_next_id("mul") == "mul2"
assert graph_id_generator.next_id("sub") == "sub1"
assert graph_id_generator.next_id("mul") == "mul1"
assert graph_id_generator.next_id("sub") == "sub2"
assert graph_id_generator.next_id("mul") == "mul2"
test/test_inputport.py
View file @ e8d99dd0
......@@ -9,42 +9,37 @@ from b_asic import Signal
@pytest.fixture
def inp_port():
return InputPort(0, None)
return InputPort(None, 0)
@pytest.fixture
def out_port():
return OutputPort(0, None)
return OutputPort(None, 0)
@pytest.fixture
def out_port2():
return OutputPort(1, None)
return OutputPort(None, 1)
@pytest.fixture
def dangling_sig():
return Signal()
@pytest.fixture
def s_w_source():
out_port = OutputPort(0, None)
def s_w_source(out_port):
return Signal(source=out_port)
@pytest.fixture
def sig_with_dest():
inp_port = InputPort(0, None)
return Signal(destination=out_port)
def sig_with_dest(inp_port):
return Signal(destination=inp_port)
@pytest.fixture
def connected_sig():
out_port = OutputPort(0, None)
inp_port = InputPort(0, None)
def connected_sig(inp_port, out_port):
return Signal(source=out_port, destination=inp_port)
def test_connect_then_disconnect(inp_port, out_port):
"""Test connect unused port to port."""
s1 = inp_port.connect(out_port)
assert inp_port.connected_ports == [out_port]
assert out_port.connected_ports == [inp_port]
assert inp_port.connected_source == out_port
assert inp_port.signals == [s1]
assert out_port.signals == [s1]
assert s1.source is out_port
......@@ -52,8 +47,7 @@ def test_connect_then_disconnect(inp_port, out_port):
inp_port.remove_signal(s1)
assert inp_port.connected_ports == []
assert out_port.connected_ports == []
assert inp_port.connected_source is None
assert inp_port.signals == []
assert out_port.signals == [s1]
assert s1.source is out_port
......@@ -62,34 +56,46 @@ def test_connect_then_disconnect(inp_port, out_port):
def test_connect_used_port_to_new_port(inp_port, out_port, out_port2):
"""Does connecting multiple ports to an inputport throw error?"""
inp_port.connect(out_port)
with pytest.raises(AssertionError):
with pytest.raises(Exception):
inp_port.connect(out_port2)
def test_add_signal_then_disconnect(inp_port, s_w_source):
"""Can signal be connected then disconnected properly?"""
inp_port.add_signal(s_w_source)
assert inp_port.connected_ports == [s_w_source.source]
assert s_w_source.source.connected_ports == [inp_port]
assert inp_port.connected_source == s_w_source.source
assert inp_port.signals == [s_w_source]
assert s_w_source.source.signals == [s_w_source]
assert s_w_source.destination is inp_port
inp_port.remove_signal(s_w_source)
assert inp_port.connected_ports == []
assert s_w_source.source.connected_ports == []
assert inp_port.connected_source is None
assert inp_port.signals == []
assert s_w_source.source.signals == [s_w_source]
assert s_w_source.destination is None
def test_connect_then_disconnect(inp_port, out_port):
"""Can port be connected and then disconnected properly?"""
inp_port.connect(out_port)
def test_set_value_length_pos_int(inp_port):
inp_port.value_length = 10
assert inp_port.value_length == 10
def test_set_value_length_zero(inp_port):
inp_port.value_length = 0
assert inp_port.value_length == 0
def test_set_value_length_neg_int(inp_port):
with pytest.raises(Exception):
inp_port.value_length = -10
def test_set_value_length_complex(inp_port):
with pytest.raises(Exception):
inp_port.value_length = (2+4j)
inp_port.disconnect(out_port)
def test_set_value_length_float(inp_port):
with pytest.raises(Exception):
inp_port.value_length = 3.2
print("outport signals:", out_port.signals, "count:", out_port.signal_count())
assert inp_port.signal_count() == 1
assert len(inp_port.connected_ports) == 0
assert out_port.signal_count() == 0
def test_set_value_length_pos_then_none(inp_port):
inp_port.value_length = 10
inp_port.value_length = None
assert inp_port.value_length is None
test/test_operation.py
View file @ e8d99dd0
from b_asic.core_operations import Constant, Addition
from b_asic.core_operations import Constant, Addition, ConstantAddition, Butterfly
from b_asic.signal import Signal
from b_asic.port import InputPort, OutputPort
import pytest
class TestTraverse:
def test_traverse_single_tree(self, operation):
"""Traverse a tree consisting of one operation."""
......@@ -20,12 +21,11 @@ class TestTraverse:
def test_traverse_type(self, large_operation_tree):
traverse = list(large_operation_tree.traverse())
assert len(list(filter(lambda type_: isinstance(type_, Addition), traverse))) == 3
assert len(list(filter(lambda type_: isinstance(type_, Constant), traverse))) == 4
assert len(
list(filter(lambda type_: isinstance(type_, Addition), traverse))) == 3
assert len(
list(filter(lambda type_: isinstance(type_, Constant), traverse))) == 4
def test_traverse_loop(self, operation_tree):
add_oper_signal = Signal()
operation_tree._output_ports[0].add_signal(add_oper_signal)
operation_tree._input_ports[0].remove_signal(add_oper_signal)
operation_tree._input_ports[0].add_signal(add_oper_signal)
assert len(list(operation_tree.traverse())) == 2
# TODO: Construct a graph that contains a loop and make sure you can traverse it properly.
assert True