diff --git a/.gitignore b/.gitignore
index 0bdf5476787722ba2007a85c8c73255d56178262..3395e18ae0404b87b2abc5759426b6ef38a49d09 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,33 +1,33 @@
-.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
diff --git a/b_asic/__init__.py b/b_asic/__init__.py
index 7e40ad52cdc51da3e1d91964ad55cfc90e12a34a..bd3574ba07b2556fae03ff8fa22deecfd2656705 100644
--- a/b_asic/__init__.py
+++ b/b_asic/__init__.py
@@ -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 *
diff --git a/b_asic/core_operations.py b/b_asic/core_operations.py
index e837d0070db95a0430671b8ae91235aa50ca829f..a1a149d787f831405558b774993b1b0ef86fe0be 100644
--- a/b_asic/core_operations.py
+++ b/b_asic/core_operations.py
@@ -4,41 +4,39 @@ TODO: More info.
"""
from numbers import Number
+from typing import Optional
from numpy import conjugate, sqrt, abs as np_abs
-from b_asic.port import InputPort, OutputPort
+
+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):
@@ -46,294 +44,215 @@ 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
-
- 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 "cdiv"
+ def evaluate(self, a):
+ return a / self.param("value")
class Butterfly(AbstractOperation):
"""Butterfly operation that returns two outputs.
@@ -341,16 +260,8 @@ class Butterfly(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), OutputPort(1, self)]
-
- if source1 is not None:
- self._input_ports[0].connect(source1)
-
- if source2 is not None:
- self._input_ports[1].connect(source2)
+ def __init__(self, src0: Optional[SignalSourceProvider] = None, 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
diff --git a/b_asic/graph_component.py b/b_asic/graph_component.py
index 1987d4491e12089fec401eedc14fb91a7274252e..52eba17c7e343842f636870d5d9a8fa694b713da 100644
--- a/b_asic/graph_component.py
+++ b/b_asic/graph_component.py
@@ -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
diff --git a/b_asic/graph_id.py b/b_asic/graph_id.py
deleted file mode 100644
index 8da6a9d4af6a1bee25125904527a2fd3a374ab90..0000000000000000000000000000000000000000
--- a/b_asic/graph_id.py
+++ /dev/null
@@ -1,26 +0,0 @@
-"""@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
diff --git a/b_asic/operation.py b/b_asic/operation.py
index 3dd761c2cd18bc97aac5e15bfd85575ed84da8f9..d644dbd3c566406be1723b18493281aa98cf6ff6 100644
--- a/b_asic/operation.py
+++ b/b_asic/operation.py
@@ -3,58 +3,89 @@ 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, Sequence, 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
-
-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
- @abstractmethod
- def evaluate_output(self, i: int, inputs: Sequence[Number]) -> Sequence[Optional[Number]]:
- """Evaluate the output port at the entered index with the entered input values and
- returns all output values that are calulated during the evaluation in a list."""
- raise NotImplementedError
-
@abstractmethod
def params(self) -> Dict[str, Optional[Any]]:
"""Get a dictionary of all parameter values."""
@@ -70,12 +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_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.
"""
@@ -83,62 +125,114 @@ 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
+ def evaluate(self, *inputs) -> Any: # pylint: disable=arguments-differ
"""Evaluate the operation and generate a list of output values given a
list of input values.
"""
raise NotImplementedError
- def evaluate_output(self, i: int, inputs: Sequence[Number]) -> Sequence[Optional[Number]]:
- eval_return = self.evaluate(*inputs)
- if isinstance(eval_return, Number):
- return [eval_return]
- elif isinstance(eval_return, (list, tuple)):
- return eval_return
- else:
- raise TypeError("Incorrect returned type from evaluate function.")
+ 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 inputs(self) -> List["InputPort"]:
+ 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()
@@ -146,36 +240,49 @@ 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 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()
@@ -185,62 +292,15 @@ 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
diff --git a/b_asic/port.py b/b_asic/port.py
index 5900afb1adab0252386a70a46d9cbc7202a99b28..4f249e3cf81d19943996e2056499a323d6c10a73 100644
--- a/b_asic/port.py
+++ b/b_asic/port.py
@@ -4,12 +4,13 @@ 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
-PortIndex = NewType("PortIndex", int)
+if TYPE_CHECKING:
+ from b_asic.operation import Operation
class Port(ABC):
@@ -20,38 +21,26 @@ class Port(ABC):
@property
@abstractmethod
- def operation(self) -> Operation:
+ def operation(self) -> "Operation":
"""Return the connected operation."""
raise NotImplementedError
@property
@abstractmethod
- def index(self) -> PortIndex:
- """Return the unique PortIndex."""
+ def index(self) -> int:
+ """Return the index of the port."""
raise NotImplementedError
@property
- @abstractmethod
- def signals(self) -> List[Signal]:
- """Return a list of all connected signals."""
- raise NotImplementedError
-
- @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
@@ -61,14 +50,6 @@ class Port(ABC):
"""
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
-
@abstractmethod
def remove_signal(self, signal: Signal) -> None:
"""Remove the signal that was entered from the Ports signals.
@@ -92,22 +73,34 @@ class AbstractPort(Port):
Handles functionality for port id and saves the connection to the parent operation.
"""
+ _operation: "Operation"
_index: int
- _operation: Operation
- def __init__(self, index: int, operation: Operation):
- self._index = index
+ def __init__(self, operation: "Operation", index: int):
self._operation = operation
+ self._index = index
@property
- def operation(self) -> Operation:
+ def operation(self) -> "Operation":
return self._operation
@property
- def index(self) -> PortIndex:
+ def index(self) -> int:
return self._index
+class SignalSourceProvider(ABC):
+ """Signal source provider interface.
+ TODO: More info.
+ """
+
+ @property
+ @abstractmethod
+ def source(self) -> "OutputPort":
+ """Get the main source port provided by this object."""
+ raise NotImplementedError
+
+
class InputPort(AbstractPort):
"""Input port.
TODO: More info.
@@ -116,110 +109,93 @@ class InputPort(AbstractPort):
_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]
-
- @property
- def value_length(self) -> Optional[int]:
- """Return the InputPorts value length."""
- return self._value_length
-
- @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]
-
- @value_length.setter
- def value_length(self, bits: Optional[int]) -> None:
- assert (isinstance(bits, int) and bits >= 0) or bits is None, \
- "Value length on input port has to be a non-negative integer or None."
- self._value_length = bits
-
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."
- # self._source_signal is set by the signal constructor.
- return Signal(port, self)
+ @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
-class OutputPort(AbstractPort):
+ def connect(self, src: SignalSourceProvider) -> 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."
+ return Signal(src.source, self) # self._source_signal is set by the signal constructor.
+
+ @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, 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()
-
- @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:
- # Signal is added to self._destination_signals in signal constructor.
- return Signal(self, port)
+ @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
\ No newline at end of file
diff --git a/b_asic/signal.py b/b_asic/signal.py
index 460bf5db9c7335adf52914610ab95381d6aed17c..67e1d0f908ba57f5d355e77794993587343e63cf 100644
--- a/b_asic/signal.py
+++ b/b_asic/signal.py
@@ -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,
+ def __init__(self, source: Optional["OutputPort"] = None, \
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
diff --git a/b_asic/signal_flow_graph.py b/b_asic/signal_flow_graph.py
index 9c08aecc40ff77b8fe90051b6ea165c0f1703b9b..a011653f4db4d85c5a9e91e0ce72d62472d658e5 100644
--- a/b_asic/signal_flow_graph.py
+++ b/b_asic/signal_flow_graph.py
@@ -3,66 +3,135 @@ B-ASIC Signal Flow Graph Module.
TODO: More info.
"""
-from typing import List, Dict, Optional, DefaultDict
+from typing import NewType, List, Iterable, Sequence, Dict, Optional, DefaultDict, Set
+from numbers import Number
from collections import defaultdict
-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):
"""Signal flow graph.
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: Sequence[Signal] = [], output_signals: Sequence[Signal] = [], \
+ inputs: Sequence[Input] = [], outputs: Sequence[Output] = [], operations: Sequence[Operation] = [], \
+ 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._components_by_id = dict()
+ self._components_by_name = defaultdict(list)
+ self._graph_id_generator = GraphIDGenerator(id_number_offset)
+ self._input_operations = []
+ self._output_operations = []
+ self._original_components_added = set()
+ self._original_input_signals = {}
+ self._original_output_signals = {}
+
+ # Setup input operations and signals.
+ for i, s in enumerate(input_signals):
+ self._input_operations.append(self._add_component_copy_unconnected(Input()))
+ self._original_input_signals[s] = i
+ for i, op in enumerate(inputs, len(input_signals)):
+ self._input_operations.append(self._add_component_copy_unconnected(op))
+ for s in op.output(0).signals:
+ self._original_input_signals[s] = i
+
+ # Setup output operations and signals.
+ for i, s in enumerate(output_signals):
+ self._output_operations.append(self._add_component_copy_unconnected(Output()))
+ self._original_output_signals[s] = i
+ for i, op in enumerate(outputs, len(output_signals)):
+ self._output_operations.append(self._add_component_copy_unconnected(op))
+ for s in op.input(0).signals:
+ self._original_output_signals[s] = i
+
+ # Search the graph inwards from each input signal.
+ for s, i in self._original_input_signals.items():
+ if s.destination is None:
+ raise ValueError(f"Input signal #{i} is missing destination in SFG")
+ if s.destination.operation not in self._original_components_added:
+ self._add_operation_copy_recursively(s.destination.operation)
+
+ # Search the graph inwards from each output signal.
+ for s, i in self._original_output_signals.items():
+ if s.source is None:
+ raise ValueError(f"Output signal #{i} is missing source in SFG")
+ if s.source.operation not in self._original_components_added:
+ self._add_operation_copy_recursively(s.source.operation)
+
+ # Search the graph outwards from each operation.
+ for op in operations:
+ if op not in self._original_components_added:
+ self._add_operation_copy_recursively(op)
- 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 = []
-
- 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()
-
- for operation in ops:
- self._add_graph_component(operation)
-
- for input_signal in input_signals:
- self._add_graph_component(input_signal)
+ @property
+ def type_name(self) -> TypeName:
+ return "sfg"
- # 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.
+ 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))
- def evaluate(self, *inputs) -> list:
- return [] # TODO: Implement
+ n = len(result)
+ return None if n == 0 else result[0] if n == 1 else result
- 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.
+ 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
- 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)
+ def split(self) -> Iterable[Operation]:
+ return filter(lambda comp: isinstance(comp, Operation), self._components_by_id.values())
- # 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
+ @property
+ def components(self) -> Iterable[GraphComponent]:
+ """Get all components of this graph."""
+ return self._components_by_id.values()
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 +140,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 +150,60 @@ class SFG(AbstractOperation):
Keyword arguments:
name: Name of the wanted object.
"""
- return self._graph_components_by_name[name]
-
- @property
- def type_name(self) -> TypeName:
- return "sfg"
+ return self._components_by_name.get(name, [])
+
+ def _add_component_copy_unconnected(self, original_comp: GraphComponent) -> GraphComponent:
+ assert original_comp not in self._original_components_added, "Tried to add duplicate SFG component"
+ self._original_components_added.add(original_comp)
+
+ new_comp = original_comp.copy_unconnected()
+ 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 _add_operation_copy_recursively(self, original_op: Operation) -> Operation:
+ # Add a copy of the operation without any connections.
+ new_op = self._add_component_copy_unconnected(original_op)
+
+ # Connect input ports.
+ for original_input_port, new_input_port in zip(original_op.inputs, new_op.inputs):
+ if original_input_port.signal_count < 1:
+ raise ValueError("Unconnected input port in SFG")
+ for original_signal in original_input_port.signals:
+ if original_signal in self._original_input_signals: # Check if the signal is one of the SFG's input signals.
+ new_signal = self._add_component_copy_unconnected(original_signal)
+ new_signal.set_destination(new_input_port)
+ new_signal.set_source(self._input_operations[self._original_input_signals[original_signal]].output(0))
+ elif original_signal not in self._original_components_added: # Only add the signal if it wasn't already added.
+ new_signal = self._add_component_copy_unconnected(original_signal)
+ new_signal.set_destination(new_input_port)
+ if original_signal.source is None:
+ raise ValueError("Dangling signal without source in SFG")
+ # Recursively add the connected operation.
+ new_connected_op = self._add_operation_copy_recursively(original_signal.source.operation)
+ new_signal.set_source(new_connected_op.output(original_signal.source.index))
+
+ # Connect output ports.
+ for original_output_port, new_output_port in zip(original_op.outputs, new_op.outputs):
+ for original_signal in original_output_port.signals:
+ if original_signal in self._original_output_signals: # Check if the signal is one of the SFG's output signals.
+ new_signal = self._add_component_copy_unconnected(original_signal)
+ new_signal.set_source(new_output_port)
+ new_signal.set_destination(self._output_operations[self._original_output_signals[original_signal]].input(0))
+ elif original_signal not in self._original_components_added: # Only add the signal if it wasn't already added.
+ new_signal = self._add_component_copy_unconnected(original_signal)
+ new_signal.set_source(new_output_port)
+ if original_signal.destination is None:
+ raise ValueError("Dangling signal without destination in SFG")
+ # Recursively add the connected operation.
+ new_connected_op = self._add_operation_copy_recursively(original_signal.destination.operation)
+ new_signal.set_destination(new_connected_op.input(original_signal.destination.index))
+
+ return new_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)
\ No newline at end of file
diff --git a/b_asic/special_operations.py b/b_asic/special_operations.py
new file mode 100644
index 0000000000000000000000000000000000000000..465c0086d0120b10e27f769a216874b2e08dd53c
--- /dev/null
+++ b/b_asic/special_operations.py
@@ -0,0 +1,54 @@
+"""@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
diff --git a/src/main.cpp b/src/main.cpp
index 75a77ef58b86cd29238205a078cec780a6ba9a36..bc4e83c69e7d331bbacfa37d8b22baec35833682 100644
--- a/src/main.cpp
+++ b/src/main.cpp
@@ -1,21 +1,21 @@
-#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
diff --git a/test/fixtures/operation_tree.py b/test/fixtures/operation_tree.py
index df3fcac35cc495d14bed06ccdfc2a3ebed25616e..94a1e42f724fdf7f14dbd13debaccc850fbbf552 100644
--- a/test/fixtures/operation_tree.py
+++ b/test/fixtures/operation_tree.py
@@ -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)))
diff --git a/test/fixtures/port.py b/test/fixtures/port.py
index 4019b3a2016aa418daeca771f9a2d8bcc4ca6652..63632ecdb3a9d81a7f27759cd7166af3163c9e94 100644
--- a/test/fixtures/port.py
+++ b/test/fixtures/port.py
@@ -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)
diff --git a/test/fixtures/signal.py b/test/fixtures/signal.py
index 7b13c9789f94c33338d08b48373391398bd9f71d..0c5692feb3203f37876e48df0ab7f2caa69c4d45 100644
--- a/test/fixtures/signal.py
+++ b/test/fixtures/signal.py
@@ -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)]
diff --git a/test/test_graph_id_generator.py b/test/test_graph_id_generator.py
index b14597eabe6c15695c5c452f69f3deeab56e36d5..b8e0cdebb7f1cc32297bacff89314244dda7cd6f 100644
--- a/test/test_graph_id_generator.py
+++ b/test/test_graph_id_generator.py
@@ -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"
diff --git a/test/test_inputport.py b/test/test_inputport.py
index a0f24e4e90cb69553a200e0a8402ac248c14d3b9..b43bf8e3d11eb3286c087c6a8bbb0b46956e51fb 100644
--- a/test/test_inputport.py
+++ b/test/test_inputport.py
@@ -9,15 +9,15 @@ 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():
@@ -39,8 +39,7 @@ 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
@@ -48,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
@@ -58,23 +56,21 @@ 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
diff --git a/test/test_operation.py b/test/test_operation.py
index 5891f3f8038bcf1aa451ff43092989ddb7bc8196..c3a05bb5a08fa443753c2bafcf2b035274098455 100644
--- a/test/test_operation.py
+++ b/test/test_operation.py
@@ -27,48 +27,5 @@ class TestTraverse:
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
-
-
-class TestEvaluateOutput:
- def test_evaluate_output_two_real_inputs(self):
- """Test evaluate_output for two real numbered inputs."""
- add1 = Addition()
-
- assert list(add1.evaluate_output(0, [1, 2])) == [3]
-
- def test_evaluate_output_addition_two_complex_inputs(self):
- """Test evaluate_output for two complex numbered inputs."""
- add1 = Addition()
-
- assert list(add1.evaluate_output(0, [1+1j, 2])) == [3+1j]
-
- def test_evaluate_output_one_real_input(self):
- """Test evaluate_output for one real numbered inputs."""
- c_add1 = ConstantAddition(5)
-
- assert list(c_add1.evaluate_output(0, [1])) == [6]
-
- def test_evaluate_output_one_complex_input(self):
- """Test evaluate_output for one complex numbered inputs."""
- c_add1 = ConstantAddition(5)
-
- assert list(c_add1.evaluate_output(0, [1+1j])) == [6+1j]
-
- def test_evaluate_output_two_real_inputs_two_outputs(self):
- """Test evaluate_output for two real inputs and two outputs."""
- bfly1 = Butterfly()
-
- assert list(bfly1.evaluate_output(0, [6, 9])) == [15, -3]
- assert list(bfly1.evaluate_output(1, [6, 9])) == [15, -3]
-
- def test_evaluate_output_two_complex_inputs_two_outputs(self):
- """Test evaluate_output for two complex inputs and two outputs."""
- bfly1 = Butterfly()
-
- assert list(bfly1.evaluate_output(0, [3+2j, 4+2j])) == [7+4j, -1]
- assert list(bfly1.evaluate_output(1, [3+2j, 4+2j])) == [7+4j, -1]
+ # TODO: Construct a graph that contains a loop and make sure you can traverse it properly.
+ assert True
diff --git a/test/test_outputport.py b/test/test_outputport.py
index deed7a1e06836600254e3903b8b45a3d05f17cbe..21f08764ac4d7f9497dc02615cce343120598959 100644
--- a/test/test_outputport.py
+++ b/test/test_outputport.py
@@ -6,75 +6,79 @@ import pytest
@pytest.fixture
def output_port():
- return OutputPort(0, None)
+ return OutputPort(None, 0)
@pytest.fixture
def input_port():
- return InputPort(0, None)
+ return InputPort(None, 0)
@pytest.fixture
def list_of_input_ports():
- return [InputPort(_, None) for _ in range(0,3)]
+ return [InputPort(None, i) for i in range(0, 3)]
class TestConnect:
def test_multiple_ports(self, output_port, list_of_input_ports):
"""Can multiple ports connect to an output port?"""
for port in list_of_input_ports:
- output_port.connect(port)
+ port.connect(output_port)
- assert output_port.signal_count() == len(list_of_input_ports)
+ assert output_port.signal_count == len(list_of_input_ports)
def test_same_port(self, output_port, list_of_input_ports):
"""Check error handing."""
- output_port.connect(list_of_input_ports[0])
- with pytest.raises(AssertionError):
- output_port.connect(list_of_input_ports[0])
+ list_of_input_ports[0].connect(output_port)
+ with pytest.raises(Exception):
+ list_of_input_ports[0].connect(output_port)
- assert output_port.signal_count() == 2
+ assert output_port.signal_count == 1
class TestAddSignal:
def test_dangling(self, output_port):
s = Signal()
output_port.add_signal(s)
- assert output_port.signal_count() == 1
-
- def test_with_destination(self, output_port, input_port):
- s = Signal(destination=input_port)
- output_port.add_signal(s)
-
- assert output_port.connected_ports == [s.destination]
+ assert output_port.signal_count == 1
+ assert output_port.signals == [s]
class TestDisconnect:
- def test_multiple_ports(self, output_port, list_of_input_ports):
+ def test_others_clear(self, output_port, list_of_input_ports):
"""Can multiple ports disconnect from OutputPort?"""
for port in list_of_input_ports:
- output_port.connect(port)
+ port.connect(output_port)
for port in list_of_input_ports:
- output_port.disconnect(port)
+ port.clear()
+
+ assert output_port.signal_count == 3
+ assert all(s.dangling() for s in output_port.signals)
- assert output_port.signal_count() == 3
- assert output_port.connected_ports == []
+ def test_self_clear(self, output_port, list_of_input_ports):
+ """Can an OutputPort disconnect from multiple ports?"""
+ for port in list_of_input_ports:
+ port.connect(output_port)
+
+ output_port.clear()
+
+ assert output_port.signal_count == 0
+ assert output_port.signals == []
class TestRemoveSignal:
def test_one_signal(self, output_port, input_port):
- s = output_port.connect(input_port)
+ s = input_port.connect(output_port)
output_port.remove_signal(s)
- assert output_port.signal_count() == 0
+ assert output_port.signal_count == 0
assert output_port.signals == []
- assert output_port.connected_ports == []
def test_multiple_signals(self, output_port, list_of_input_ports):
"""Can multiple signals disconnect from OutputPort?"""
sigs = []
for port in list_of_input_ports:
- sigs.append(output_port.connect(port))
+ sigs.append(port.connect(output_port))
- for sig in sigs:
- output_port.remove_signal(sig)
+ for s in sigs:
+ output_port.remove_signal(s)
- assert output_port.signal_count() == 0
+ assert output_port.signal_count == 0
assert output_port.signals == []
diff --git a/test/test_sfg.py b/test/test_sfg.py
new file mode 100644
index 0000000000000000000000000000000000000000..d3daf2e96db0cd87350b148e0969febb2397a0fd
--- /dev/null
+++ b/test/test_sfg.py
@@ -0,0 +1,32 @@
+from b_asic import SFG
+from b_asic.signal import Signal
+from b_asic.core_operations import Addition, Constant
+from b_asic.special_operations import Input, Output
+
+class TestConstructor:
+ def test_outputs_construction(self, operation_tree):
+ outp = Output(operation_tree)
+ sfg = SFG(outputs=[outp])
+
+ assert len(list(sfg.components)) == 7
+ assert sfg.input_count == 0
+ assert sfg.output_count == 1
+
+ def test_signals_construction(self, operation_tree):
+ outs = Signal(source=operation_tree.output(0))
+ sfg = SFG(output_signals=[outs])
+
+ assert len(list(sfg.components)) == 7
+ assert sfg.input_count == 0
+ assert sfg.output_count == 1
+
+ def test_operations_construction(self, operation_tree):
+ sfg1 = SFG(operations=[operation_tree])
+ sfg2 = SFG(operations=[operation_tree.input(1).signals[0].source.operation])
+
+ assert len(list(sfg1.components)) == 5
+ assert len(list(sfg2.components)) == 5
+ assert sfg1.input_count == 0
+ assert sfg2.input_count == 0
+ assert sfg1.output_count == 0
+ assert sfg2.output_count == 0
diff --git a/test/test_signal.py b/test/test_signal.py
index ab07eb778ddb693bfc9cfabf6aeb7804038312d5..9a45086a99e55089c9e25100cdd56399ca46a5cc 100644
--- a/test/test_signal.py
+++ b/test/test_signal.py
@@ -8,8 +8,8 @@ from b_asic.signal import Signal
import pytest
def test_signal_creation_and_disconnction_and_connection_changing():
- in_port = InputPort(0, None)
- out_port = OutputPort(1, None)
+ in_port = InputPort(None, 0)
+ out_port = OutputPort(None, 1)
s = Signal(out_port, in_port)
assert in_port.signals == [s]
@@ -17,7 +17,7 @@ def test_signal_creation_and_disconnction_and_connection_changing():
assert s.source is out_port
assert s.destination is in_port
- in_port1 = InputPort(0, None)
+ in_port1 = InputPort(None, 0)
s.set_destination(in_port1)
assert in_port.signals == []
@@ -40,7 +40,7 @@ def test_signal_creation_and_disconnction_and_connection_changing():
assert s.source is None
assert s.destination is None
- out_port1 = OutputPort(0, None)
+ out_port1 = OutputPort(None, 0)
s.set_source(out_port1)
assert out_port1.signals == [s]