Skip to content
Snippets Groups Projects
Commit 022ef035 authored by Ivar Härnqvist's avatar Ivar Härnqvist Committed by Angus Lothian
Browse files

Resolve "Create SFG"

parent 3ace2303
No related branches found
No related tags found
2 merge requests!67WIP: B-ASIC version 1.0.0 hotfix,!65B-ASIC version 1.0.0
Hide whitespace changes
Inline Side-by-side
Showing
with 750 additions and 691 deletions
.gitignore
+ 32
− 32
View file @ 022ef035
.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
+ 1
− 1
View file @ 022ef035
......@@ -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
+ 87
− 176
View file @ 022ef035
......@@ -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
......
b_asic/graph_component.py
+ 11
− 0
View file @ 022ef035
......@@ -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
+ 0
− 26
View file @ 3ace2303
"""@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
+ 170
− 110
View file @ 022ef035
......@@ -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
b_asic/port.py
+ 81
− 105
View file @ 022ef035
......@@ -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
b_asic/signal.py
+ 28
− 34
View file @ 022ef035
......@@ -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
b_asic/signal_flow_graph.py
+ 170
− 52
View file @ 022ef035
......@@ -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
b_asic/special_operations.py 0 → 100644
+ 54
− 0
View file @ 022ef035
"""@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
+ 20
− 20
View file @ 022ef035
#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
+ 6
− 34
View file @ 022ef035
......@@ -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
+ 2
− 2
View file @ 022ef035
......@@ -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
+ 1
− 1
View file @ 022ef035
......@@ -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_graph_id_generator.py
+ 9
− 9
View file @ 022ef035
......@@ -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
+ 8
− 12
View file @ 022ef035
......@@ -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
......
test/test_operation.py
+ 2
− 45
View file @ 022ef035
......@@ -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
test/test_outputport.py
+ 32
− 28
View file @ 022ef035
......@@ -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 == []
test/test_sfg.py 0 → 100644
+ 32
− 0
View file @ 022ef035
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
test/test_signal.py
+ 4
− 4
View file @ 022ef035
......@@ -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]
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment