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

Improve typing

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