Skip to content
Snippets Groups Projects
Commit 3e22e222 authored by Oscar Gustafsson's avatar Oscar Gustafsson :bicyclist:
Browse files

Merge branch 'operationtypechecking' into 'master' 

Improve typing

See merge request !109
parents 689208f9 dde6f31d
No related branches found
No related tags found
1 merge request!109Improve typing
Pipeline #88177 passed
Stage: test
Stage: deploy
Hide whitespace changes
Inline Side-by-side
b_asic/core_operations.py
+ 45
− 45
View file @ 3e22e222
...@@ -26,19 +26,19 @@ class Constant(AbstractOperation): ...@@ -26,19 +26,19 @@ class Constant(AbstractOperation):
_execution_time = 0 _execution_time = 0
def __init__(self, value: Number = 0, name: Name = ""): def __init__(self, value: Number = 0, name: Name = Name("")):
"""Construct a Constant operation with the given value.""" """Construct a Constant operation with the given value."""
super().__init__( super().__init__(
input_count=0, input_count=0,
output_count=1, output_count=1,
name=name, name=Name(name),
latency_offsets={"out0": 0}, latency_offsets={"out0": 0},
) )
self.set_param("value", value) self.set_param("value", value)
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "c" return TypeName("c")
def evaluate(self): def evaluate(self):
return self.param("value") return self.param("value")
...@@ -67,7 +67,7 @@ class Addition(AbstractOperation): ...@@ -67,7 +67,7 @@ class Addition(AbstractOperation):
self, self,
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
src1: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None,
name: Name = "", name: Name = Name(""),
latency: Optional[int] = None, latency: Optional[int] = None,
latency_offsets: Optional[Dict[str, int]] = None, latency_offsets: Optional[Dict[str, int]] = None,
): ):
...@@ -75,7 +75,7 @@ class Addition(AbstractOperation): ...@@ -75,7 +75,7 @@ class Addition(AbstractOperation):
super().__init__( super().__init__(
input_count=2, input_count=2,
output_count=1, output_count=1,
name=name, name=Name(name),
input_sources=[src0, src1], input_sources=[src0, src1],
latency=latency, latency=latency,
latency_offsets=latency_offsets, latency_offsets=latency_offsets,
...@@ -83,7 +83,7 @@ class Addition(AbstractOperation): ...@@ -83,7 +83,7 @@ class Addition(AbstractOperation):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "add" return TypeName("add")
def evaluate(self, a, b): def evaluate(self, a, b):
return a + b return a + b
...@@ -102,7 +102,7 @@ class Subtraction(AbstractOperation): ...@@ -102,7 +102,7 @@ class Subtraction(AbstractOperation):
self, self,
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
src1: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None,
name: Name = "", name: Name = Name(""),
latency: Optional[int] = None, latency: Optional[int] = None,
latency_offsets: Optional[Dict[str, int]] = None, latency_offsets: Optional[Dict[str, int]] = None,
): ):
...@@ -110,7 +110,7 @@ class Subtraction(AbstractOperation): ...@@ -110,7 +110,7 @@ class Subtraction(AbstractOperation):
super().__init__( super().__init__(
input_count=2, input_count=2,
output_count=1, output_count=1,
name=name, name=Name(name),
input_sources=[src0, src1], input_sources=[src0, src1],
latency=latency, latency=latency,
latency_offsets=latency_offsets, latency_offsets=latency_offsets,
...@@ -118,7 +118,7 @@ class Subtraction(AbstractOperation): ...@@ -118,7 +118,7 @@ class Subtraction(AbstractOperation):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "sub" return TypeName("sub")
def evaluate(self, a, b): def evaluate(self, a, b):
return a - b return a - b
...@@ -139,7 +139,7 @@ class AddSub(AbstractOperation): ...@@ -139,7 +139,7 @@ class AddSub(AbstractOperation):
is_add: bool = True, is_add: bool = True,
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
src1: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None,
name: Name = "", name: Name = Name(""),
latency: Optional[int] = None, latency: Optional[int] = None,
latency_offsets: Optional[Dict[str, int]] = None, latency_offsets: Optional[Dict[str, int]] = None,
): ):
...@@ -147,7 +147,7 @@ class AddSub(AbstractOperation): ...@@ -147,7 +147,7 @@ class AddSub(AbstractOperation):
super().__init__( super().__init__(
input_count=2, input_count=2,
output_count=1, output_count=1,
name=name, name=Name(name),
input_sources=[src0, src1], input_sources=[src0, src1],
latency=latency, latency=latency,
latency_offsets=latency_offsets, latency_offsets=latency_offsets,
...@@ -156,7 +156,7 @@ class AddSub(AbstractOperation): ...@@ -156,7 +156,7 @@ class AddSub(AbstractOperation):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "addsub" return TypeName("addsub")
def evaluate(self, a, b): def evaluate(self, a, b):
return a + b if self.is_add else a - b return a + b if self.is_add else a - b
...@@ -185,7 +185,7 @@ class Multiplication(AbstractOperation): ...@@ -185,7 +185,7 @@ class Multiplication(AbstractOperation):
self, self,
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
src1: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None,
name: Name = "", name: Name = Name(""),
latency: Optional[int] = None, latency: Optional[int] = None,
latency_offsets: Optional[Dict[str, int]] = None, latency_offsets: Optional[Dict[str, int]] = None,
): ):
...@@ -193,7 +193,7 @@ class Multiplication(AbstractOperation): ...@@ -193,7 +193,7 @@ class Multiplication(AbstractOperation):
super().__init__( super().__init__(
input_count=2, input_count=2,
output_count=1, output_count=1,
name=name, name=Name(name),
input_sources=[src0, src1], input_sources=[src0, src1],
latency=latency, latency=latency,
latency_offsets=latency_offsets, latency_offsets=latency_offsets,
...@@ -201,7 +201,7 @@ class Multiplication(AbstractOperation): ...@@ -201,7 +201,7 @@ class Multiplication(AbstractOperation):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "mul" return TypeName("mul")
def evaluate(self, a, b): def evaluate(self, a, b):
return a * b return a * b
...@@ -220,7 +220,7 @@ class Division(AbstractOperation): ...@@ -220,7 +220,7 @@ class Division(AbstractOperation):
self, self,
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
src1: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None,
name: Name = "", name: Name = Name(""),
latency: Optional[int] = None, latency: Optional[int] = None,
latency_offsets: Optional[Dict[str, int]] = None, latency_offsets: Optional[Dict[str, int]] = None,
): ):
...@@ -228,7 +228,7 @@ class Division(AbstractOperation): ...@@ -228,7 +228,7 @@ class Division(AbstractOperation):
super().__init__( super().__init__(
input_count=2, input_count=2,
output_count=1, output_count=1,
name=name, name=Name(name),
input_sources=[src0, src1], input_sources=[src0, src1],
latency=latency, latency=latency,
latency_offsets=latency_offsets, latency_offsets=latency_offsets,
...@@ -236,7 +236,7 @@ class Division(AbstractOperation): ...@@ -236,7 +236,7 @@ class Division(AbstractOperation):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "div" return TypeName("div")
def evaluate(self, a, b): def evaluate(self, a, b):
return a / b return a / b
...@@ -256,7 +256,7 @@ class Min(AbstractOperation): ...@@ -256,7 +256,7 @@ class Min(AbstractOperation):
self, self,
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
src1: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None,
name: Name = "", name: Name = Name(""),
latency: Optional[int] = None, latency: Optional[int] = None,
latency_offsets: Optional[Dict[str, int]] = None, latency_offsets: Optional[Dict[str, int]] = None,
): ):
...@@ -264,7 +264,7 @@ class Min(AbstractOperation): ...@@ -264,7 +264,7 @@ class Min(AbstractOperation):
super().__init__( super().__init__(
input_count=2, input_count=2,
output_count=1, output_count=1,
name=name, name=Name(name),
input_sources=[src0, src1], input_sources=[src0, src1],
latency=latency, latency=latency,
latency_offsets=latency_offsets, latency_offsets=latency_offsets,
...@@ -272,7 +272,7 @@ class Min(AbstractOperation): ...@@ -272,7 +272,7 @@ class Min(AbstractOperation):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "min" return TypeName("min")
def evaluate(self, a, b): def evaluate(self, a, b):
if isinstance(a, complex) or isinstance(b, complex): if isinstance(a, complex) or isinstance(b, complex):
...@@ -296,7 +296,7 @@ class Max(AbstractOperation): ...@@ -296,7 +296,7 @@ class Max(AbstractOperation):
self, self,
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
src1: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None,
name: Name = "", name: Name = Name(""),
latency: Optional[int] = None, latency: Optional[int] = None,
latency_offsets: Optional[Dict[str, int]] = None, latency_offsets: Optional[Dict[str, int]] = None,
): ):
...@@ -304,7 +304,7 @@ class Max(AbstractOperation): ...@@ -304,7 +304,7 @@ class Max(AbstractOperation):
super().__init__( super().__init__(
input_count=2, input_count=2,
output_count=1, output_count=1,
name=name, name=Name(name),
input_sources=[src0, src1], input_sources=[src0, src1],
latency=latency, latency=latency,
latency_offsets=latency_offsets, latency_offsets=latency_offsets,
...@@ -312,7 +312,7 @@ class Max(AbstractOperation): ...@@ -312,7 +312,7 @@ class Max(AbstractOperation):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "max" return TypeName("max")
def evaluate(self, a, b): def evaluate(self, a, b):
if isinstance(a, complex) or isinstance(b, complex): if isinstance(a, complex) or isinstance(b, complex):
...@@ -334,7 +334,7 @@ class SquareRoot(AbstractOperation): ...@@ -334,7 +334,7 @@ class SquareRoot(AbstractOperation):
def __init__( def __init__(
self, self,
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
name: Name = "", name: Name = Name(""),
latency: Optional[int] = None, latency: Optional[int] = None,
latency_offsets: Optional[Dict[str, int]] = None, latency_offsets: Optional[Dict[str, int]] = None,
): ):
...@@ -342,7 +342,7 @@ class SquareRoot(AbstractOperation): ...@@ -342,7 +342,7 @@ class SquareRoot(AbstractOperation):
super().__init__( super().__init__(
input_count=1, input_count=1,
output_count=1, output_count=1,
name=name, name=Name(name),
input_sources=[src0], input_sources=[src0],
latency=latency, latency=latency,
latency_offsets=latency_offsets, latency_offsets=latency_offsets,
...@@ -350,7 +350,7 @@ class SquareRoot(AbstractOperation): ...@@ -350,7 +350,7 @@ class SquareRoot(AbstractOperation):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "sqrt" return TypeName("sqrt")
def evaluate(self, a): def evaluate(self, a):
return sqrt(complex(a)) return sqrt(complex(a))
...@@ -368,7 +368,7 @@ class ComplexConjugate(AbstractOperation): ...@@ -368,7 +368,7 @@ class ComplexConjugate(AbstractOperation):
def __init__( def __init__(
self, self,
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
name: Name = "", name: Name = Name(""),
latency: Optional[int] = None, latency: Optional[int] = None,
latency_offsets: Optional[Dict[str, int]] = None, latency_offsets: Optional[Dict[str, int]] = None,
): ):
...@@ -376,7 +376,7 @@ class ComplexConjugate(AbstractOperation): ...@@ -376,7 +376,7 @@ class ComplexConjugate(AbstractOperation):
super().__init__( super().__init__(
input_count=1, input_count=1,
output_count=1, output_count=1,
name=name, name=Name(name),
input_sources=[src0], input_sources=[src0],
latency=latency, latency=latency,
latency_offsets=latency_offsets, latency_offsets=latency_offsets,
...@@ -384,7 +384,7 @@ class ComplexConjugate(AbstractOperation): ...@@ -384,7 +384,7 @@ class ComplexConjugate(AbstractOperation):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "conj" return TypeName("conj")
def evaluate(self, a): def evaluate(self, a):
return conjugate(a) return conjugate(a)
...@@ -402,7 +402,7 @@ class Absolute(AbstractOperation): ...@@ -402,7 +402,7 @@ class Absolute(AbstractOperation):
def __init__( def __init__(
self, self,
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
name: Name = "", name: Name = Name(""),
latency: Optional[int] = None, latency: Optional[int] = None,
latency_offsets: Optional[Dict[str, int]] = None, latency_offsets: Optional[Dict[str, int]] = None,
): ):
...@@ -410,7 +410,7 @@ class Absolute(AbstractOperation): ...@@ -410,7 +410,7 @@ class Absolute(AbstractOperation):
super().__init__( super().__init__(
input_count=1, input_count=1,
output_count=1, output_count=1,
name=name, name=Name(name),
input_sources=[src0], input_sources=[src0],
latency=latency, latency=latency,
latency_offsets=latency_offsets, latency_offsets=latency_offsets,
...@@ -418,7 +418,7 @@ class Absolute(AbstractOperation): ...@@ -418,7 +418,7 @@ class Absolute(AbstractOperation):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "abs" return TypeName("abs")
def evaluate(self, a): def evaluate(self, a):
return np_abs(a) return np_abs(a)
...@@ -437,7 +437,7 @@ class ConstantMultiplication(AbstractOperation): ...@@ -437,7 +437,7 @@ class ConstantMultiplication(AbstractOperation):
self, self,
value: Number = 0, value: Number = 0,
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
name: Name = "", name: Name = Name(""),
latency: Optional[int] = None, latency: Optional[int] = None,
latency_offsets: Optional[Dict[str, int]] = None, latency_offsets: Optional[Dict[str, int]] = None,
): ):
...@@ -446,7 +446,7 @@ class ConstantMultiplication(AbstractOperation): ...@@ -446,7 +446,7 @@ class ConstantMultiplication(AbstractOperation):
super().__init__( super().__init__(
input_count=1, input_count=1,
output_count=1, output_count=1,
name=name, name=Name(name),
input_sources=[src0], input_sources=[src0],
latency=latency, latency=latency,
latency_offsets=latency_offsets, latency_offsets=latency_offsets,
...@@ -455,7 +455,7 @@ class ConstantMultiplication(AbstractOperation): ...@@ -455,7 +455,7 @@ class ConstantMultiplication(AbstractOperation):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "cmul" return TypeName("cmul")
def evaluate(self, a): def evaluate(self, a):
return a * self.param("value") return a * self.param("value")
...@@ -486,7 +486,7 @@ class Butterfly(AbstractOperation): ...@@ -486,7 +486,7 @@ class Butterfly(AbstractOperation):
self, self,
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
src1: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None,
name: Name = "", name: Name = Name(""),
latency: Optional[int] = None, latency: Optional[int] = None,
latency_offsets: Optional[Dict[str, int]] = None, latency_offsets: Optional[Dict[str, int]] = None,
): ):
...@@ -494,7 +494,7 @@ class Butterfly(AbstractOperation): ...@@ -494,7 +494,7 @@ class Butterfly(AbstractOperation):
super().__init__( super().__init__(
input_count=2, input_count=2,
output_count=2, output_count=2,
name=name, name=Name(name),
input_sources=[src0, src1], input_sources=[src0, src1],
latency=latency, latency=latency,
latency_offsets=latency_offsets, latency_offsets=latency_offsets,
...@@ -502,7 +502,7 @@ class Butterfly(AbstractOperation): ...@@ -502,7 +502,7 @@ class Butterfly(AbstractOperation):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "bfly" return TypeName("bfly")
def evaluate(self, a, b): def evaluate(self, a, b):
return a + b, a - b return a + b, a - b
...@@ -523,7 +523,7 @@ class MAD(AbstractOperation): ...@@ -523,7 +523,7 @@ class MAD(AbstractOperation):
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
src1: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None,
src2: Optional[SignalSourceProvider] = None, src2: Optional[SignalSourceProvider] = None,
name: Name = "", name: Name = Name(""),
latency: Optional[int] = None, latency: Optional[int] = None,
latency_offsets: Optional[Dict[str, int]] = None, latency_offsets: Optional[Dict[str, int]] = None,
): ):
...@@ -531,7 +531,7 @@ class MAD(AbstractOperation): ...@@ -531,7 +531,7 @@ class MAD(AbstractOperation):
super().__init__( super().__init__(
input_count=3, input_count=3,
output_count=1, output_count=1,
name=name, name=Name(name),
input_sources=[src0, src1, src2], input_sources=[src0, src1, src2],
latency=latency, latency=latency,
latency_offsets=latency_offsets, latency_offsets=latency_offsets,
...@@ -539,7 +539,7 @@ class MAD(AbstractOperation): ...@@ -539,7 +539,7 @@ class MAD(AbstractOperation):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "mad" return TypeName("mad")
def evaluate(self, a, b, c): def evaluate(self, a, b, c):
return a * b + c return a * b + c
...@@ -558,7 +558,7 @@ class SymmetricTwoportAdaptor(AbstractOperation): ...@@ -558,7 +558,7 @@ class SymmetricTwoportAdaptor(AbstractOperation):
value: Number = 0, value: Number = 0,
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
src1: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None,
name: Name = "", name: Name = Name(""),
latency: Optional[int] = None, latency: Optional[int] = None,
latency_offsets: Optional[Dict[str, int]] = None, latency_offsets: Optional[Dict[str, int]] = None,
): ):
...@@ -566,7 +566,7 @@ class SymmetricTwoportAdaptor(AbstractOperation): ...@@ -566,7 +566,7 @@ class SymmetricTwoportAdaptor(AbstractOperation):
super().__init__( super().__init__(
input_count=2, input_count=2,
output_count=2, output_count=2,
name=name, name=Name(name),
input_sources=[src0, src1], input_sources=[src0, src1],
latency=latency, latency=latency,
latency_offsets=latency_offsets, latency_offsets=latency_offsets,
...@@ -575,7 +575,7 @@ class SymmetricTwoportAdaptor(AbstractOperation): ...@@ -575,7 +575,7 @@ class SymmetricTwoportAdaptor(AbstractOperation):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "sym2p" return TypeName("sym2p")
def evaluate(self, a, b): def evaluate(self, a, b):
tmp = self.value * (b - a) tmp = self.value * (b - a)
......
b_asic/graph_component.py
+ 3
− 3
View file @ 3e22e222
...@@ -113,10 +113,10 @@ class AbstractGraphComponent(GraphComponent): ...@@ -113,10 +113,10 @@ class AbstractGraphComponent(GraphComponent):
_graph_id: GraphID _graph_id: GraphID
_parameters: Dict[str, Any] _parameters: Dict[str, Any]
def __init__(self, name: Name = ""): def __init__(self, name: Name = Name("")):
"""Construct a graph component.""" """Construct a graph component."""
self._name = name self._name = Name(name)
self._graph_id = "" self._graph_id = GraphID("")
self._parameters = {} self._parameters = {}
def __str__(self) -> str: def __str__(self) -> str:
......
b_asic/operation.py
+ 65
− 51
View file @ 3e22e222
...@@ -18,15 +18,33 @@ from typing import ( ...@@ -18,15 +18,33 @@ from typing import (
NewType, NewType,
Optional, Optional,
Sequence, Sequence,
Set,
Tuple, Tuple,
Union, Union,
TYPE_CHECKING,
) )
from b_asic.graph_component import AbstractGraphComponent, GraphComponent, Name from b_asic.graph_component import (
AbstractGraphComponent,
GraphComponent,
GraphID,
Name,
)
from b_asic.port import InputPort, OutputPort, SignalSourceProvider from b_asic.port import InputPort, OutputPort, SignalSourceProvider
from b_asic.signal import Signal from b_asic.signal import Signal
if TYPE_CHECKING:
# Conditionally imported to avoid circular imports
from b_asic.core_operations import (
Addition,
Subtraction,
Multiplication,
ConstantMultiplication,
Division,
)
from b_asic.signal_flow_graph import SFG
ResultKey = NewType("ResultKey", str) ResultKey = NewType("ResultKey", str)
ResultMap = Mapping[ResultKey, Optional[Number]] ResultMap = Mapping[ResultKey, Optional[Number]]
MutableResultMap = MutableMapping[ResultKey, Optional[Number]] MutableResultMap = MutableMapping[ResultKey, Optional[Number]]
...@@ -63,7 +81,9 @@ class Operation(GraphComponent, SignalSourceProvider): ...@@ -63,7 +81,9 @@ class Operation(GraphComponent, SignalSourceProvider):
raise NotImplementedError raise NotImplementedError
@abstractmethod @abstractmethod
def __sub__(self, src: Union[SignalSourceProvider, Number]) -> "Subtraction": def __sub__(
self, src: Union[SignalSourceProvider, Number]
) -> "Subtraction":
""" """
Overloads the subtraction operator to make it return a new Subtraction operation Overloads the subtraction operator to make it return a new Subtraction operation
object that is connected to the self and other objects. object that is connected to the self and other objects.
...@@ -305,10 +325,9 @@ class Operation(GraphComponent, SignalSourceProvider): ...@@ -305,10 +325,9 @@ class Operation(GraphComponent, SignalSourceProvider):
@property @property
@abstractmethod @abstractmethod
def latency_offsets(self) -> Sequence[Sequence[int]]: def latency_offsets(self) -> Dict[str, int]:
"""Get a nested list with all the operations ports latency-offsets, the first list contains the """
latency-offsets of the operations input ports, the second list contains the latency-offsets of Get a dictionary with all the operations ports latency-offsets.
the operations output ports.
""" """
raise NotImplementedError raise NotImplementedError
...@@ -332,7 +351,7 @@ class Operation(GraphComponent, SignalSourceProvider): ...@@ -332,7 +351,7 @@ class Operation(GraphComponent, SignalSourceProvider):
@property @property
@abstractmethod @abstractmethod
def execution_time(self) -> int: def execution_time(self) -> Optional[int]:
""" """
Get the execution time of the operation, which is the time it takes before the Get the execution time of the operation, which is the time it takes before the
processing element implementing the operation can be reused for starting another operation. processing element implementing the operation can be reused for starting another operation.
...@@ -351,7 +370,7 @@ class Operation(GraphComponent, SignalSourceProvider): ...@@ -351,7 +370,7 @@ class Operation(GraphComponent, SignalSourceProvider):
@abstractmethod @abstractmethod
def get_plot_coordinates( def get_plot_coordinates(
self, self,
) -> Tuple[List[List[Number]], List[List[Number]]]: ) -> Tuple[List[List[float]], List[List[float]]]:
""" """
Get a tuple constaining coordinates for the two polygons outlining Get a tuple constaining coordinates for the two polygons outlining
the latency and execution time of the operation. the latency and execution time of the operation.
...@@ -362,7 +381,7 @@ class Operation(GraphComponent, SignalSourceProvider): ...@@ -362,7 +381,7 @@ class Operation(GraphComponent, SignalSourceProvider):
@abstractmethod @abstractmethod
def get_io_coordinates( def get_io_coordinates(
self, self,
) -> Tuple[List[List[Number]], List[List[Number]]]: ) -> Tuple[List[List[float]], List[List[float]]]:
""" """
Get a tuple constaining coordinates for inputs and outputs, respectively. Get a tuple constaining coordinates for inputs and outputs, respectively.
These maps to the polygons and are corresponding to a start time of 0 These maps to the polygons and are corresponding to a start time of 0
...@@ -387,7 +406,7 @@ class AbstractOperation(Operation, AbstractGraphComponent): ...@@ -387,7 +406,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
self, self,
input_count: int, input_count: int,
output_count: int, output_count: int,
name: Name = "", name: Name = Name(""),
input_sources: Optional[ input_sources: Optional[
Sequence[Optional[SignalSourceProvider]] Sequence[Optional[SignalSourceProvider]]
] = None, ] = None,
...@@ -404,7 +423,7 @@ class AbstractOperation(Operation, AbstractGraphComponent): ...@@ -404,7 +423,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
The latency offsets may also be specified to be initialized. The latency offsets may also be specified to be initialized.
""" """
super().__init__(name) super().__init__(Name(name))
self._input_ports = [InputPort(self, i) for i in range(input_count)] self._input_ports = [InputPort(self, i) for i in range(input_count)]
self._output_ports = [OutputPort(self, i) for i in range(output_count)] self._output_ports = [OutputPort(self, i) for i in range(output_count)]
...@@ -421,26 +440,18 @@ class AbstractOperation(Operation, AbstractGraphComponent): ...@@ -421,26 +440,18 @@ class AbstractOperation(Operation, AbstractGraphComponent):
if src is not None: if src is not None:
self._input_ports[i].connect(src.source) self._input_ports[i].connect(src.source)
ports_without_latency_offset = set(
(
[f"in{i}" for i in range(self.input_count)]
+ [f"out{i}" for i in range(self.output_count)]
)
)
if latency_offsets is not None:
self.set_latency_offsets(latency_offsets)
if latency is not None: if latency is not None:
# Set the latency of the rest of ports with no latency_offset. # Set the latency for all ports initially.
if latency < 0: if latency < 0:
raise ValueError("Latency cannot be negative") raise ValueError("Latency cannot be negative")
for inp in self.inputs: for inp in self.inputs:
if inp.latency_offset is None: inp.latency_offset = 0
inp.latency_offset = 0
for outp in self.outputs: for outp in self.outputs:
if outp.latency_offset is None: outp.latency_offset = latency
outp.latency_offset = latency
# Set specific latency_offsets
if latency_offsets is not None:
self.set_latency_offsets(latency_offsets)
self._execution_time = execution_time self._execution_time = execution_time
...@@ -550,41 +561,41 @@ class AbstractOperation(Operation, AbstractGraphComponent): ...@@ -550,41 +561,41 @@ class AbstractOperation(Operation, AbstractGraphComponent):
def __str__(self) -> str: def __str__(self) -> str:
"""Get a string representation of this operation.""" """Get a string representation of this operation."""
inputs_dict = {} inputs_dict = {}
for i, port in enumerate(self.inputs): for i, inport in enumerate(self.inputs):
if port.signal_count == 0: if inport.signal_count == 0:
inputs_dict[i] = "-" inputs_dict[i] = "-"
break break
dict_ele = [] dict_ele = []
for signal in port.signals: for signal in inport.signals:
if signal.source: if signal.source:
if signal.source.operation.graph_id: if signal.source.operation.graph_id:
dict_ele.append(signal.source.operation.graph_id) dict_ele.append(signal.source.operation.graph_id)
else: else:
dict_ele.append("no_id") dict_ele.append(GraphID("no_id"))
else: else:
if signal.graph_id: if signal.graph_id:
dict_ele.append(signal.graph_id) dict_ele.append(signal.graph_id)
else: else:
dict_ele.append("no_id") dict_ele.append(GraphID("no_id"))
inputs_dict[i] = dict_ele inputs_dict[i] = dict_ele
outputs_dict = {} outputs_dict = {}
for i, port in enumerate(self.outputs): for i, outport in enumerate(self.outputs):
if port.signal_count == 0: if outport.signal_count == 0:
outputs_dict[i] = "-" outputs_dict[i] = "-"
break break
dict_ele = [] dict_ele = []
for signal in port.signals: for signal in outport.signals:
if signal.destination: if signal.destination:
if signal.destination.operation.graph_id: if signal.destination.operation.graph_id:
dict_ele.append(signal.destination.operation.graph_id) dict_ele.append(signal.destination.operation.graph_id)
else: else:
dict_ele.append("no_id") dict_ele.append(GraphID("no_id"))
else: else:
if signal.graph_id: if signal.graph_id:
dict_ele.append(signal.graph_id) dict_ele.append(signal.graph_id)
else: else:
dict_ele.append("no_id") dict_ele.append(GraphID("no_id"))
outputs_dict[i] = dict_ele outputs_dict[i] = dict_ele
return ( return (
...@@ -638,7 +649,7 @@ class AbstractOperation(Operation, AbstractGraphComponent): ...@@ -638,7 +649,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
key += str(index) key += str(index)
elif not key: elif not key:
key = str(index) key = str(index)
return key return ResultKey(key)
def current_output( def current_output(
self, index: int, delays: Optional[DelayMap] = None, prefix: str = "" self, index: int, delays: Optional[DelayMap] = None, prefix: str = ""
...@@ -871,7 +882,7 @@ class AbstractOperation(Operation, AbstractGraphComponent): ...@@ -871,7 +882,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
) )
@property @property
def latency_offsets(self) -> Sequence[Sequence[int]]: def latency_offsets(self) -> Dict[str, int]:
latency_offsets = {} latency_offsets = {}
for i, inp in enumerate(self.inputs): for i, inp in enumerate(self.inputs):
...@@ -914,7 +925,7 @@ class AbstractOperation(Operation, AbstractGraphComponent): ...@@ -914,7 +925,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
) )
@property @property
def execution_time(self) -> Union[int, None]: def execution_time(self) -> Optional[int]:
"""Execution time of operation.""" """Execution time of operation."""
return self._execution_time return self._execution_time
...@@ -924,13 +935,13 @@ class AbstractOperation(Operation, AbstractGraphComponent): ...@@ -924,13 +935,13 @@ class AbstractOperation(Operation, AbstractGraphComponent):
raise ValueError("Execution time cannot be negative") raise ValueError("Execution time cannot be negative")
self._execution_time = execution_time self._execution_time = execution_time
def _increase_time_resolution(self, factor: int): def _increase_time_resolution(self, factor: int) -> None:
if self._execution_time is not None: if self._execution_time is not None:
self._execution_time *= factor self._execution_time *= factor
for port in [*self.inputs, *self.outputs]: for port in [*self.inputs, *self.outputs]:
port.latency_offset *= factor port.latency_offset *= factor
def _decrease_time_resolution(self, factor: int): def _decrease_time_resolution(self, factor: int) -> None:
if self._execution_time is not None: if self._execution_time is not None:
self._execution_time = self._execution_time // factor self._execution_time = self._execution_time // factor
for port in [*self.inputs, *self.outputs]: for port in [*self.inputs, *self.outputs]:
...@@ -938,37 +949,39 @@ class AbstractOperation(Operation, AbstractGraphComponent): ...@@ -938,37 +949,39 @@ class AbstractOperation(Operation, AbstractGraphComponent):
def get_plot_coordinates( def get_plot_coordinates(
self, self,
) -> Tuple[List[List[Number]], List[List[Number]]]: ) -> Tuple[List[List[float]], List[List[float]]]:
# Doc-string inherited
return ( return (
self._get_plot_coordinates_for_latency(), self._get_plot_coordinates_for_latency(),
self._get_plot_coordinates_for_execution_time(), self._get_plot_coordinates_for_execution_time(),
) )
def _get_plot_coordinates_for_execution_time(self) -> List[List[Number]]: def _get_plot_coordinates_for_execution_time(self) -> List[List[float]]:
# Always a rectangle, but easier if coordinates are returned # Always a rectangle, but easier if coordinates are returned
if self._execution_time is None: execution_time = self._execution_time # Copy for type checking
if execution_time is None:
return [] return []
return [ return [
[0, 0], [0, 0],
[0, 1], [0, 1],
[self.execution_time, 1], [execution_time, 1],
[self.execution_time, 0], [execution_time, 0],
[0, 0], [0, 0],
] ]
def _get_plot_coordinates_for_latency(self) -> List[List[Number]]: def _get_plot_coordinates_for_latency(self) -> List[List[float]]:
# Points for latency polygon # Points for latency polygon
latency = [] latency = []
# Remember starting point # Remember starting point
start_point = [self.inputs[0].latency_offset, 0] start_point = [self.inputs[0].latency_offset, 0]
num_in = len(self.inputs) num_in = self.input_count
latency.append(start_point) latency.append(start_point)
for k in range(1, num_in): for k in range(1, num_in):
latency.append([self.inputs[k - 1].latency_offset, k / num_in]) latency.append([self.inputs[k - 1].latency_offset, k / num_in])
latency.append([self.inputs[k].latency_offset, k / num_in]) latency.append([self.inputs[k].latency_offset, k / num_in])
latency.append([self.inputs[num_in - 1].latency_offset, 1]) latency.append([self.inputs[num_in - 1].latency_offset, 1])
num_out = len(self.outputs) num_out = self.output_count
latency.append([self.outputs[num_out - 1].latency_offset, 1]) latency.append([self.outputs[num_out - 1].latency_offset, 1])
for k in reversed(range(1, num_out)): for k in reversed(range(1, num_out)):
latency.append([self.outputs[k].latency_offset, k / num_out]) latency.append([self.outputs[k].latency_offset, k / num_out])
...@@ -981,7 +994,8 @@ class AbstractOperation(Operation, AbstractGraphComponent): ...@@ -981,7 +994,8 @@ class AbstractOperation(Operation, AbstractGraphComponent):
def get_io_coordinates( def get_io_coordinates(
self, self,
) -> Tuple[List[List[Number]], List[List[Number]]]: ) -> Tuple[List[List[float]], List[List[float]]]:
# Doc-string inherited
input_coords = [ input_coords = [
[ [
self.inputs[k].latency_offset, self.inputs[k].latency_offset,
......
b_asic/port.py
+ 6
− 4
View file @ 3e22e222
...@@ -42,7 +42,7 @@ class Port(ABC): ...@@ -42,7 +42,7 @@ class Port(ABC):
@property @property
@abstractmethod @abstractmethod
def latency_offset(self) -> int: def latency_offset(self) -> Optional[int]:
"""Get the latency_offset of the port.""" """Get the latency_offset of the port."""
raise NotImplementedError raise NotImplementedError
...@@ -126,7 +126,7 @@ class AbstractPort(Port): ...@@ -126,7 +126,7 @@ class AbstractPort(Port):
return self._index return self._index
@property @property
def latency_offset(self) -> int: def latency_offset(self) -> Optional[int]:
return self._latency_offset return self._latency_offset
@latency_offset.setter @latency_offset.setter
...@@ -199,7 +199,9 @@ class InputPort(AbstractPort): ...@@ -199,7 +199,9 @@ class InputPort(AbstractPort):
None if self._source_signal is None else self._source_signal.source None if self._source_signal is None else self._source_signal.source
) )
def connect(self, src: SignalSourceProvider, name: Name = "") -> Signal: def connect(
self, src: SignalSourceProvider, name: Name = Name("")
) -> Signal:
""" """
Connect the provided signal source to this input port by creating a new signal. Connect the provided signal source to this input port by creating a new signal.
Returns the new signal. Returns the new signal.
...@@ -207,7 +209,7 @@ class InputPort(AbstractPort): ...@@ -207,7 +209,7 @@ class InputPort(AbstractPort):
if self._source_signal is not None: if self._source_signal is not None:
raise ValueError("Cannot connect already connected input port.") raise ValueError("Cannot connect already connected input port.")
# self._source_signal is set by the signal constructor. # self._source_signal is set by the signal constructor.
return Signal(source=src.source, destination=self, name=name) return Signal(source=src.source, destination=self, name=Name(name))
def __lshift__(self, src: SignalSourceProvider) -> Signal: def __lshift__(self, src: SignalSourceProvider) -> Signal:
""" """
......
b_asic/schedule.py
+ 1
− 1
View file @ 3e22e222
...@@ -136,7 +136,7 @@ class Schedule: ...@@ -136,7 +136,7 @@ class Schedule:
def _backward_slacks( def _backward_slacks(
self, op_id: GraphID self, op_id: GraphID
) -> Dict["OutputPort", Dict["Signal", int]]: ) -> Dict[OutputPort, Dict[Signal, int]]:
ret = {} ret = {}
start_time = self._start_times[op_id] start_time = self._start_times[op_id]
op = self._sfg.find_by_id(op_id) op = self._sfg.find_by_id(op_id)
......
b_asic/signal.py
+ 3
− 3
View file @ 3e22e222
...@@ -27,10 +27,10 @@ class Signal(AbstractGraphComponent): ...@@ -27,10 +27,10 @@ class Signal(AbstractGraphComponent):
source: Optional["OutputPort"] = None, source: Optional["OutputPort"] = None,
destination: Optional["InputPort"] = None, destination: Optional["InputPort"] = None,
bits: Optional[int] = None, bits: Optional[int] = None,
name: Name = "", name: Name = Name(""),
): ):
"""Construct a Signal.""" """Construct a Signal."""
super().__init__(name) super().__init__(Name(name))
self._source = None self._source = None
self._destination = None self._destination = None
if source is not None: if source is not None:
...@@ -41,7 +41,7 @@ class Signal(AbstractGraphComponent): ...@@ -41,7 +41,7 @@ class Signal(AbstractGraphComponent):
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "s" return TypeName("s")
@property @property
def neighbors(self) -> Iterable[GraphComponent]: def neighbors(self) -> Iterable[GraphComponent]:
......
b_asic/signal_flow_graph.py
+ 7
− 7
View file @ 3e22e222
...@@ -63,12 +63,12 @@ class GraphIDGenerator: ...@@ -63,12 +63,12 @@ class GraphIDGenerator:
while new_id in used_ids: while new_id in used_ids:
self._next_id_number[type_name] += 1 self._next_id_number[type_name] += 1
new_id = type_name + str(self._next_id_number[type_name]) new_id = type_name + str(self._next_id_number[type_name])
return new_id return GraphID(new_id)
@property @property
def id_number_offset(self) -> GraphIDNumber: def id_number_offset(self) -> GraphIDNumber:
"""Get the graph id number offset of this generator.""" """Get the graph id number offset of this generator."""
return ( return GraphIDNumber(
self._next_id_number.default_factory() self._next_id_number.default_factory()
) # pylint: disable=not-callable ) # pylint: disable=not-callable
...@@ -300,7 +300,7 @@ class SFG(AbstractOperation): ...@@ -300,7 +300,7 @@ class SFG(AbstractOperation):
return string_io.getvalue() return string_io.getvalue()
def __call__( def __call__(
self, *src: Optional[SignalSourceProvider], name: Name = "" self, *src: Optional[SignalSourceProvider], name: Name = Name("")
) -> "SFG": ) -> "SFG":
""" """
Get a new independent SFG instance that is identical to this SFG Get a new independent SFG instance that is identical to this SFG
...@@ -310,13 +310,13 @@ class SFG(AbstractOperation): ...@@ -310,13 +310,13 @@ class SFG(AbstractOperation):
inputs=self._input_operations, inputs=self._input_operations,
outputs=self._output_operations, outputs=self._output_operations,
id_number_offset=self.id_number_offset, id_number_offset=self.id_number_offset,
name=name, name=Name(name),
input_sources=src if src else None, input_sources=src if src else None,
) )
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "sfg" return TypeName("sfg")
def evaluate(self, *args): def evaluate(self, *args):
result = self.evaluate_outputs(args) result = self.evaluate_outputs(args)
...@@ -424,14 +424,14 @@ class SFG(AbstractOperation): ...@@ -424,14 +424,14 @@ class SFG(AbstractOperation):
return True return True
@property @property
def input_operations(self) -> Sequence[Operation]: def input_operations(self) -> Iterable[Operation]:
""" """
Get the internal input operations in the same order as their respective input ports. Get the internal input operations in the same order as their respective input ports.
""" """
return self._input_operations return self._input_operations
@property @property
def output_operations(self) -> Sequence[Operation]: def output_operations(self) -> Iterable[Operation]:
""" """
Get the internal output operations in the same order as their respective output ports. Get the internal output operations in the same order as their respective output ports.
""" """
......
b_asic/special_operations.py
+ 21
− 13
View file @ 3e22e222
...@@ -14,7 +14,6 @@ from b_asic.operation import ( ...@@ -14,7 +14,6 @@ from b_asic.operation import (
DelayMap, DelayMap,
MutableDelayMap, MutableDelayMap,
MutableResultMap, MutableResultMap,
ResultKey,
) )
from b_asic.port import SignalSourceProvider from b_asic.port import SignalSourceProvider
...@@ -29,19 +28,19 @@ class Input(AbstractOperation): ...@@ -29,19 +28,19 @@ class Input(AbstractOperation):
_execution_time = 0 _execution_time = 0
def __init__(self, name: Name = ""): def __init__(self, name: Name = Name("")):
"""Construct an Input operation.""" """Construct an Input operation."""
super().__init__( super().__init__(
input_count=0, input_count=0,
output_count=1, output_count=1,
name=name, name=Name(name),
latency_offsets={"out0": 0}, latency_offsets={"out0": 0},
) )
self.set_param("value", 0) self.set_param("value", 0)
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "in" return TypeName("in")
def evaluate(self): def evaluate(self):
return self.param("value") return self.param("value")
...@@ -58,7 +57,8 @@ class Input(AbstractOperation): ...@@ -58,7 +57,8 @@ class Input(AbstractOperation):
def get_plot_coordinates( def get_plot_coordinates(
self, self,
) -> Tuple[List[List[Number]], List[List[Number]]]: ) -> Tuple[List[List[float]], List[List[float]]]:
# Doc-string inherited
return ( return (
[ [
[-0.5, 0], [-0.5, 0],
...@@ -80,7 +80,8 @@ class Input(AbstractOperation): ...@@ -80,7 +80,8 @@ class Input(AbstractOperation):
def get_io_coordinates( def get_io_coordinates(
self, self,
) -> Tuple[List[List[Number]], List[List[Number]]]: ) -> Tuple[List[List[float]], List[List[float]]]:
# Doc-string inherited
return ([], [[0, 0.5]]) return ([], [[0, 0.5]])
...@@ -96,27 +97,30 @@ class Output(AbstractOperation): ...@@ -96,27 +97,30 @@ class Output(AbstractOperation):
_execution_time = 0 _execution_time = 0
def __init__( def __init__(
self, src0: Optional[SignalSourceProvider] = None, name: Name = "" self,
src0: Optional[SignalSourceProvider] = None,
name: Name = Name(""),
): ):
"""Construct an Output operation.""" """Construct an Output operation."""
super().__init__( super().__init__(
input_count=1, input_count=1,
output_count=0, output_count=0,
name=name, name=Name(name),
input_sources=[src0], input_sources=[src0],
latency_offsets={"in0": 0}, latency_offsets={"in0": 0},
) )
@classmethod @classmethod
def type_name(cls) -> TypeName: def type_name(cls) -> TypeName:
return "out" return TypeName("out")
def evaluate(self, _): def evaluate(self, _):
return None return None
def get_plot_coordinates( def get_plot_coordinates(
self, self,
) -> Tuple[List[List[Number]], List[List[Number]]]: ) -> Tuple[List[List[float]], List[List[float]]]:
# Doc-string inherited
return ( return (
[[0, 0], [0, 1], [0.25, 1], [0.5, 0.5], [0.25, 0], [0, 0]], [[0, 0], [0, 1], [0.25, 1], [0.5, 0.5], [0.25, 0], [0, 0]],
[[0, 0], [0, 1], [0.25, 1], [0.5, 0.5], [0.25, 0], [0, 0]], [[0, 0], [0, 1], [0.25, 1], [0.5, 0.5], [0.25, 0], [0, 0]],
...@@ -124,7 +128,8 @@ class Output(AbstractOperation): ...@@ -124,7 +128,8 @@ class Output(AbstractOperation):
def get_io_coordinates( def get_io_coordinates(
self, self,
) -> Tuple[List[List[Number]], List[List[Number]]]: ) -> Tuple[List[List[float]], List[List[float]]]:
# Doc-string inherited
return ([[0, 0.5]], []) return ([[0, 0.5]], [])
...@@ -140,11 +145,14 @@ class Delay(AbstractOperation): ...@@ -140,11 +145,14 @@ class Delay(AbstractOperation):
self, self,
src0: Optional[SignalSourceProvider] = None, src0: Optional[SignalSourceProvider] = None,
initial_value: Number = 0, initial_value: Number = 0,
name: Name = "", name: Name = Name(""),
): ):
"""Construct a Delay operation.""" """Construct a Delay operation."""
super().__init__( super().__init__(
input_count=1, output_count=1, name=name, input_sources=[src0] input_count=1,
output_count=1,
name=Name(name),
input_sources=[src0],
) )
self.set_param("initial_value", initial_value) self.set_param("initial_value", initial_value)
......
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