From dde6f31d0279be18a7218b63ab2ef60f50dd2c55 Mon Sep 17 00:00:00 2001
From: Oscar Gustafsson <oscar.gustafsson@gmail.com>
Date: Thu, 26 Jan 2023 09:58:30 +0100
Subject: [PATCH] Improve typing
---
b_asic/core_operations.py | 90 +++++++++++++--------------
b_asic/graph_component.py | 6 +-
b_asic/operation.py | 116 ++++++++++++++++++++---------------
b_asic/port.py | 10 +--
b_asic/schedule.py | 2 +-
b_asic/signal.py | 6 +-
b_asic/signal_flow_graph.py | 14 ++---
b_asic/special_operations.py | 34 ++++++----
8 files changed, 151 insertions(+), 127 deletions(-)
diff --git a/b_asic/core_operations.py b/b_asic/core_operations.py
index a6984423..6041228a 100644
--- a/b_asic/core_operations.py
+++ b/b_asic/core_operations.py
@@ -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)
diff --git a/b_asic/graph_component.py b/b_asic/graph_component.py
index 89f7785c..28deab5a 100644
--- a/b_asic/graph_component.py
+++ b/b_asic/graph_component.py
@@ -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:
diff --git a/b_asic/operation.py b/b_asic/operation.py
index 36392e13..ff03277e 100644
--- a/b_asic/operation.py
+++ b/b_asic/operation.py
@@ -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,
diff --git a/b_asic/port.py b/b_asic/port.py
index 0ea0b28f..39dd4d98 100644
--- a/b_asic/port.py
+++ b/b_asic/port.py
@@ -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:
"""
diff --git a/b_asic/schedule.py b/b_asic/schedule.py
index 4db660c5..acaa2db7 100644
--- a/b_asic/schedule.py
+++ b/b_asic/schedule.py
@@ -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)
diff --git a/b_asic/signal.py b/b_asic/signal.py
index 0ec3cfa6..b9b80383 100644
--- a/b_asic/signal.py
+++ b/b_asic/signal.py
@@ -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]:
diff --git a/b_asic/signal_flow_graph.py b/b_asic/signal_flow_graph.py
index 55aed0ed..3adf217e 100644
--- a/b_asic/signal_flow_graph.py
+++ b/b_asic/signal_flow_graph.py
@@ -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.
"""
diff --git a/b_asic/special_operations.py b/b_asic/special_operations.py
index 02a80dac..b63e112d 100644
--- a/b_asic/special_operations.py
+++ b/b_asic/special_operations.py
@@ -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)
--
GitLab