From 3780986f330c2544a02b1650844f4fe7d22ef024 Mon Sep 17 00:00:00 2001
From: Oscar Gustafsson <oscar.gustafsson@gmail.com>
Date: Fri, 2 Jun 2023 16:49:59 +0200
Subject: [PATCH] Replace << connect operator with <<= and overload shift
---
b_asic/architecture.py | 6 +-
b_asic/operation.py | 12 +-
b_asic/port.py | 28 +-
b_asic/sfg_generators.py | 6 +-
docs_sphinx/index.rst | 6 +-
examples/connectmultiplesfgs.py | 4 +-
examples/firstorderiirfilter.py | 4 +-
examples/folding_example_with_architecture.py | 2 +-
examples/lwdfallpass.py | 3 +-
examples/schedulingexample.py | 2 +-
examples/thirdorderblwdf.py | 2 +-
test/fixtures/signal_flow_graph.py | 6 +-
test/test_operation.py | 23 ++
test/test_schedule.py | 2 +-
test/test_sfg.py | 245 ++++++++----------
15 files changed, 178 insertions(+), 173 deletions(-)
diff --git a/b_asic/architecture.py b/b_asic/architecture.py
index 8daf5391..2638cb62 100644
--- a/b_asic/architecture.py
+++ b/b_asic/architecture.py
@@ -203,6 +203,10 @@ class Resource(HardwareBlock):
def _info(self):
return ""
+ @property
+ def _color(self):
+ raise NotImplementedError
+
@property
def schedule_time(self) -> int:
# doc-string inherited
@@ -304,7 +308,7 @@ class Resource(HardwareBlock):
"""
if isinstance(proc, OperatorProcess):
# operation_type marks OperatorProcess associated operation.
- if not isinstance(proc._operation, self.operation_type):
+ if not isinstance(proc.operation, self.operation_type):
raise TypeError(f"{proc} not of type {self.operation_type}")
else:
# operation_type is MemoryVariable or PlainMemoryVariable
diff --git a/b_asic/operation.py b/b_asic/operation.py
index 4e6edd5f..1c31f788 100644
--- a/b_asic/operation.py
+++ b/b_asic/operation.py
@@ -54,11 +54,10 @@ class Operation(GraphComponent, SignalSourceProvider):
"""
@abstractmethod
- def __lshift__(self, src: SignalSourceProvider) -> Signal:
+ def __ilshift__(self, src: SignalSourceProvider) -> "Operation":
"""
- Overload the left shift operator to make it connect the provided signal source
- to this operation's input, assuming it has exactly 1 input port.
- Returns the new signal.
+ Overload the inline left shift operator to make it connect the provided signal
+ source to this operation's input, assuming it has exactly one input port.
"""
raise NotImplementedError
@@ -541,14 +540,15 @@ class AbstractOperation(Operation, AbstractGraphComponent):
"""
raise NotImplementedError
- def __lshift__(self, src: SignalSourceProvider) -> Signal:
+ def __ilshift__(self, src: SignalSourceProvider) -> "Operation":
if self.input_count != 1:
diff = "more" if self.input_count > 1 else "less"
raise TypeError(
f"{self.__class__.__name__} cannot be used as a destination"
f" because it has {diff} than 1 input"
)
- return self.input(0).connect(src)
+ self.input(0).connect(src)
+ return self
def __str__(self) -> str:
"""Get a string representation of this operation."""
diff --git a/b_asic/port.py b/b_asic/port.py
index 0542f09e..8c7e0c21 100644
--- a/b_asic/port.py
+++ b/b_asic/port.py
@@ -18,8 +18,11 @@ if TYPE_CHECKING:
Addition,
ConstantMultiplication,
Division,
+ LeftShift,
Multiplication,
Reciprocal,
+ RightShift,
+ Shift,
Subtraction,
)
from b_asic.operation import Operation
@@ -247,6 +250,24 @@ class SignalSourceProvider(ABC):
return Division(Constant(src), self)
return Division(src, self)
+ def __lshift__(self, src: int) -> Union["LeftShift", "Shift"]:
+ from b_asic.core_operations import LeftShift, Shift
+
+ if not isinstance(src, int):
+ raise TypeError("Can only shift with an int")
+ if src >= 0:
+ return LeftShift(src, self)
+ return Shift(src, self)
+
+ def __rshift__(self, src: int) -> Union["RightShift", "Shift"]:
+ from b_asic.core_operations import RightShift, Shift
+
+ if not isinstance(src, int):
+ raise TypeError("Can only shift with an int")
+ if src >= 0:
+ return RightShift(src, self)
+ return Shift(-src, self)
+
class InputPort(AbstractPort):
"""
@@ -311,12 +332,13 @@ class InputPort(AbstractPort):
# self._source_signal is set by the signal constructor.
return Signal(source=src.source, destination=self, name=Name(name))
- def __lshift__(self, src: SignalSourceProvider) -> Signal:
+ def __ilshift__(self, src: SignalSourceProvider) -> "InputPort":
"""
- Overloads the left shift operator to make it connect the provided
+ Overloads the inline left shift operator to make it connect the provided
signal source to this input port. Returns the new signal.
"""
- return self.connect(src)
+ self.connect(src)
+ return self
class OutputPort(AbstractPort, SignalSourceProvider):
diff --git a/b_asic/sfg_generators.py b/b_asic/sfg_generators.py
index e912d10b..8682db1b 100644
--- a/b_asic/sfg_generators.py
+++ b/b_asic/sfg_generators.py
@@ -117,7 +117,7 @@ def wdf_allpass(
signal_out = Signal(adaptor1.output(0))
else:
signal_out = Signal(delay2)
- output << signal_out
+ output <<= signal_out
return SFG([input_op], [output], name=Name(name))
@@ -184,7 +184,7 @@ def direct_form_fir(
if i < taps - 1:
prev_delay = Delay(prev_delay)
- output << prev_add
+ output <<= prev_add
return SFG([input_op], [output], name=Name(name))
@@ -251,6 +251,6 @@ def transposed_direct_form_fir(
if i < taps - 1:
prev_delay = Delay(tmp_add)
- output << tmp_add
+ output <<= tmp_add
return SFG([input_op], [output], name=Name(name))
diff --git a/docs_sphinx/index.rst b/docs_sphinx/index.rst
index 1fcfdf32..bcbb9c1e 100644
--- a/docs_sphinx/index.rst
+++ b/docs_sphinx/index.rst
@@ -3,8 +3,8 @@
You can adapt this file completely to your liking, but it should at least
contain the root `toctree` directive.
-Welcome to B-ASIC's documentation!
-==================================
+B-ASIC's documentation
+======================
B-ASIC is a toolbox for Python 3 that simplifies implementing application-specific
circuits, primarily aimed at signal processing algorithms for both standard-cell and
@@ -24,7 +24,7 @@ The development of B-ASIC happens at
It is not yet fully functional, but several parts of the design flow works,
while others are missing/buggy. The goal is to have a working design path from
-algorithm downto a HDL-description of a custom architecture. Once it becomes a
+algorithm down to a HDL-description of a custom architecture. Once it becomes a
bit more mature, we expect to make it available on pypi and conda-forge so
that it will becomes easier to access.
diff --git a/examples/connectmultiplesfgs.py b/examples/connectmultiplesfgs.py
index 4ab897b4..a28f28fa 100644
--- a/examples/connectmultiplesfgs.py
+++ b/examples/connectmultiplesfgs.py
@@ -23,8 +23,8 @@ allpass1 = wdf_allpass([0.2, 0.5])
allpass2 = wdf_allpass([-0.5, 0.2, 0.5])
in_lwdf = Input()
-allpass1 << in_lwdf
-allpass2 << in_lwdf
+allpass1 <<= in_lwdf
+allpass2 <<= in_lwdf
out_lwdf = Output((allpass1 + allpass2) * 0.5)
# Create SFG of LWDF with two internal SFGs
diff --git a/examples/firstorderiirfilter.py b/examples/firstorderiirfilter.py
index d34cd5de..4dea6dcf 100644
--- a/examples/firstorderiirfilter.py
+++ b/examples/firstorderiirfilter.py
@@ -40,9 +40,9 @@ b1.input(0).connect(delay)
# graph, e.g., for recursive algorithms. In this example, we could not connect the
# output of the delay as that was not yet available.
#
-# There is also a shorthand form to connect signals using the ``<<`` operator:
+# There is also a shorthand form to connect signals using the ``<<=`` operator:
-delay << first_addition
+delay <<= first_addition
# %%
# Naturally, it is also possible to write expressions when instantiating operations:
diff --git a/examples/folding_example_with_architecture.py b/examples/folding_example_with_architecture.py
index 85a7fc65..289b9d1c 100644
--- a/examples/folding_example_with_architecture.py
+++ b/examples/folding_example_with_architecture.py
@@ -30,7 +30,7 @@ d = ConstantMultiplication(0.6, T2, "d")
add2 = a + c
add1 = in1 + add2
add3 = b + d
-T1 << add1
+T1 <<= add1
out1 = Output(add1 + add3, "OUT")
sfg = SFG(inputs=[in1], outputs=[out1], name="Bi-quad folding example")
diff --git a/examples/lwdfallpass.py b/examples/lwdfallpass.py
index 15609f6c..281856fe 100644
--- a/examples/lwdfallpass.py
+++ b/examples/lwdfallpass.py
@@ -1,5 +1,4 @@
#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
"""
================================
LWDF first-order allpass section
@@ -19,7 +18,7 @@ d0 = Delay()
adaptor0 = SymmetricTwoportAdaptor(
0.5, in0, d0, latency_offsets={"in0": 0, "in1": 1, "out0": 5, "out1": 6}
)
-d0 << adaptor0.output(1)
+d0 <<= adaptor0.output(1)
out0 = Output(adaptor0.output(0))
adaptor0.execution_time = 2
sfg = SFG([in0], [out0])
diff --git a/examples/schedulingexample.py b/examples/schedulingexample.py
index e0e82b56..31663194 100644
--- a/examples/schedulingexample.py
+++ b/examples/schedulingexample.py
@@ -25,7 +25,7 @@ node7 = node6 + node4
out = Output(node7)
node5 = 0.75 * node4
node3 = node2 + node5
-node4 << node3
+node4 <<= node3
sfg = SFG([node1], [out], name="Scheduling example")
# %%
diff --git a/examples/thirdorderblwdf.py b/examples/thirdorderblwdf.py
index df19a1ed..fc289e24 100644
--- a/examples/thirdorderblwdf.py
+++ b/examples/thirdorderblwdf.py
@@ -20,7 +20,7 @@ D0 = Delay(in0)
D1 = Delay()
D2 = Delay(D1)
s = SymmetricTwoportAdaptor(-0.375, in0, D2)
-D1 << s.output(1)
+D1 <<= s.output(1)
a = s.output(0) + D0
out0 = Output(a, "y")
diff --git a/test/fixtures/signal_flow_graph.py b/test/fixtures/signal_flow_graph.py
index e8d4f5b4..173c4953 100644
--- a/test/fixtures/signal_flow_graph.py
+++ b/test/fixtures/signal_flow_graph.py
@@ -145,7 +145,7 @@ def sfg_accumulator():
data_in = Input()
reset = Input()
t = Delay()
- t << (t + data_in) * (1 - reset)
+ t <<= (t + data_in) * (1 - reset)
data_out = Output(t)
return SFG(inputs=[data_in, reset], outputs=[data_out])
@@ -164,7 +164,7 @@ def sfg_simple_accumulator():
in1 = Input()
t1 = Delay()
add1 = in1 + t1
- t1 << add1
+ t1 <<= add1
out1 = Output(add1)
return SFG(inputs=[in1], outputs=[out1])
@@ -330,5 +330,5 @@ def sfg_direct_form_iir_lp_filter():
top_node = d0 * b1 + d1 * b2 + x
d0.input(0).connect(top_node)
d1.input(0).connect(d0)
- y << a1 * d0 + a2 * d1 + a0 * top_node
+ y <<= a1 * d0 + a2 * d1 + a0 * top_node
return SFG(inputs=[x], outputs=[y], name='Direct Form 2 IIR Lowpass filter')
diff --git a/test/test_operation.py b/test/test_operation.py
index 69bfda46..dc7ad0b1 100644
--- a/test/test_operation.py
+++ b/test/test_operation.py
@@ -12,8 +12,11 @@ from b_asic import (
Constant,
ConstantMultiplication,
Division,
+ LeftShift,
Multiplication,
Reciprocal,
+ RightShift,
+ Shift,
SquareRoot,
Subtraction,
)
@@ -116,6 +119,26 @@ class TestOperationOverloading:
assert isinstance(div4, Reciprocal)
assert div4.input(0).signals == div3.output(0).signals
+ def test_shift_overload(self):
+ """Tests multiplication overloading for both operation and number argument."""
+ add1 = Addition(None, None, "add1")
+
+ ls1 = add1 << 2
+ assert isinstance(ls1, LeftShift)
+ assert ls1.input(0).signals == add1.output(0).signals
+
+ ls2 = ls1 << -2
+ assert isinstance(ls2, Shift)
+ assert ls2.input(0).signals == ls1.output(0).signals
+
+ rs1 = ls2 >> 2
+ assert isinstance(rs1, RightShift)
+ assert rs1.input(0).signals == ls2.output(0).signals
+
+ rs2 = rs1 >> -2
+ assert isinstance(rs2, Shift)
+ assert rs2.input(0).signals == rs1.output(0).signals
+
class TestTraverse:
def test_traverse_single_tree(self, operation):
diff --git a/test/test_schedule.py b/test/test_schedule.py
index cfa66db0..c9e6c038 100644
--- a/test/test_schedule.py
+++ b/test/test_schedule.py
@@ -334,7 +334,7 @@ class TestRescheduling:
d = Delay()
a = d + in0
out0 = Output(a)
- d << a
+ d <<= a
sfg = SFG([in0], [out0])
sfg.set_latency_of_type(Addition.type_name(), 1)
schedule = Schedule(sfg, cyclic=True)
diff --git a/test/test_sfg.py b/test/test_sfg.py
index 2f8c0455..2258081c 100644
--- a/test/test_sfg.py
+++ b/test/test_sfg.py
@@ -504,54 +504,40 @@ class TestGetPrecedenceList:
# Cached precedence list
assert len(precedence_sfg_delays._precedence_list) == 7
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[0]
- ]
- ) == {"IN1", "T1", "T2"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[0]
+ } == {"IN1", "T1", "T2"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[1]
- ]
- ) == {"C0", "B1", "B2", "A1", "A2"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[1]
+ } == {"C0", "B1", "B2", "A1", "A2"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[2]
- ]
- ) == {"ADD2", "ADD3"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[2]
+ } == {"ADD2", "ADD3"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[3]
- ]
- ) == {"ADD1"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[3]
+ } == {"ADD1"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[4]
- ]
- ) == {"Q1"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[4]
+ } == {"Q1"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[5]
- ]
- ) == {"A0"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[5]
+ } == {"A0"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[6]
- ]
- ) == {"ADD4"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[6]
+ } == {"ADD4"}
# Trigger cache
precedence_list = precedence_sfg_delays.get_precedence_list()
@@ -565,54 +551,40 @@ class TestGetPrecedenceList:
assert len(precedence_list) == 7
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[0]
- ]
- ) == {"IN1", "T1", "CONST1"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[0]
+ } == {"IN1", "T1", "CONST1"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[1]
- ]
- ) == {"C0", "B1", "B2", "A1", "A2"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[1]
+ } == {"C0", "B1", "B2", "A1", "A2"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[2]
- ]
- ) == {"ADD2", "ADD3"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[2]
+ } == {"ADD2", "ADD3"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[3]
- ]
- ) == {"ADD1"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[3]
+ } == {"ADD1"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[4]
- ]
- ) == {"Q1"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[4]
+ } == {"Q1"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[5]
- ]
- ) == {"A0"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[5]
+ } == {"A0"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[6]
- ]
- ) == {"BFLY1.0", "BFLY1.1"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[6]
+ } == {"BFLY1.0", "BFLY1.1"}
def test_precedence_multiple_outputs_same_precedence(
self, sfg_two_inputs_two_outputs
@@ -635,26 +607,20 @@ class TestGetPrecedenceList:
assert len(precedence_list) == 3
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[0]
- ]
- ) == {"IN1", "IN2"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[0]
+ } == {"IN1", "IN2"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[1]
- ]
- ) == {"CMUL1"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[1]
+ } == {"CMUL1"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[2]
- ]
- ) == {"NESTED_SFG.0", "NESTED_SFG.1"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[2]
+ } == {"NESTED_SFG.0", "NESTED_SFG.1"}
def test_precedence_sfg_multiple_outputs_different_precedences(
self, sfg_two_inputs_two_outputs_independent
@@ -676,26 +642,20 @@ class TestGetPrecedenceList:
assert len(precedence_list) == 3
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[0]
- ]
- ) == {"IN1", "IN2"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[0]
+ } == {"IN1", "IN2"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[1]
- ]
- ) == {"CMUL1"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[1]
+ } == {"CMUL1"}
- assert set(
- [
- port.operation.key(port.index, port.operation.name)
- for port in precedence_list[2]
- ]
- ) == {"NESTED_SFG.0", "NESTED_SFG.1"}
+ assert {
+ port.operation.key(port.index, port.operation.name)
+ for port in precedence_list[2]
+ } == {"NESTED_SFG.0", "NESTED_SFG.1"}
class TestPrintPrecedence:
@@ -924,8 +884,7 @@ class TestConnectExternalSignalsToComponentsMultipleComp:
return SFG(inputs=[inp1, inp2], outputs=[out1])
def test_connect_external_signals_to_components_many_op(self, large_operation_tree):
- """Replaces an sfg component in a larger SFG with several component operations
- """
+ """Replace an sfg component in a larger SFG with several component operations"""
inp1 = Input("INP1")
inp2 = Input("INP2")
inp3 = Input("INP3")
@@ -963,8 +922,8 @@ class TestConnectExternalSignalsToComponentsMultipleComp:
in2 = Input()
output = Output(c1_sfg + c2_sfg)
- c1_sfg << in1
- c2_sfg << in2
+ c1_sfg <<= in1
+ c2_sfg <<= in2
sfg = SFG([in1, in2], [output])
assert not sfg.find_by_type_name(ConstantMultiplication.type_name())
@@ -1030,31 +989,29 @@ class TestRemove:
def test_remove_single_input_outputs(self, sfg_simple_filter):
new_sfg = sfg_simple_filter.remove_operation("cmul1")
- assert set(
+ assert {
op.name
for op in sfg_simple_filter.find_by_name("T1")[0].subsequent_operations
- ) == {"CMUL1", "OUT1"}
- assert set(
+ } == {"CMUL1", "OUT1"}
+ assert {
op.name for op in new_sfg.find_by_name("T1")[0].subsequent_operations
- ) == {"ADD1", "OUT1"}
+ } == {"ADD1", "OUT1"}
- assert set(
+ assert {
op.name
for op in sfg_simple_filter.find_by_name("ADD1")[0].preceding_operations
- ) == {"CMUL1", "IN1"}
- assert set(
+ } == {"CMUL1", "IN1"}
+ assert {
op.name for op in new_sfg.find_by_name("ADD1")[0].preceding_operations
- ) == {"T1", "IN1"}
+ } == {"T1", "IN1"}
- assert "S1" in set(
- [
- sig.name
- for sig in sfg_simple_filter.find_by_name("T1")[0].output(0).signals
- ]
- )
- assert "S2" in set(
- [sig.name for sig in new_sfg.find_by_name("T1")[0].output(0).signals]
- )
+ assert "S1" in {
+ sig.name
+ for sig in sfg_simple_filter.find_by_name("T1")[0].output(0).signals
+ }
+ assert "S2" in {
+ sig.name for sig in new_sfg.find_by_name("T1")[0].output(0).signals
+ }
def test_remove_multiple_inputs_outputs(self, butterfly_operation_tree):
out1 = Output(butterfly_operation_tree.output(0), "OUT1")
@@ -1109,7 +1066,7 @@ class TestRemove:
assert sfg_source_1.operation.name == "bfly2"
assert new_sfg_source_1.operation.name == "bfly3"
- assert "bfly2" not in set(op.name for op in new_sfg.operations)
+ assert "bfly2" not in {op.name for op in new_sfg.operations}
def remove_different_number_inputs_outputs(self, sfg_simple_filter):
with pytest.raises(ValueError):
@@ -1134,7 +1091,7 @@ class TestSaveLoadSFG:
assert path.exists(path_)
- with open(path_, "r") as file_obj:
+ with open(path_) as file_obj:
assert file_obj.read() == result
remove(path_)
@@ -1149,7 +1106,7 @@ class TestSaveLoadSFG:
assert path.exists(path_)
- with open(path_, "r") as file_obj:
+ with open(path_) as file_obj:
assert file_obj.read() == result
remove(path_)
--
GitLab