requirements.txt

@@ -2,3 +2,4 @@ numpy
 qtpy
 graphviz>=0.19
 matplotlib
+setuptools_scm[toml]>=6.2

requirements_doc.txt

 sphinx
 furo
 numpydoc
+sphinx-gallery
+scipy

test/fixtures/schedule.py

@@ -13,5 +13,5 @@ def secondorder_iir_schedule(precedence_sfg_delays):
         ConstantMultiplication.type_name(), 3
     )
 
-    schedule = Schedule(precedence_sfg_delays, scheduling_alg="ASAP")
+    schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
     return schedule

test/test_core_operations.py

@@ -12,6 +12,7 @@ from b_asic import (
     Max,
     Min,
     Multiplication,
+    Reciprocal,
     SquareRoot,
     Subtraction,
     SymmetricTwoportAdaptor,
@@ -261,6 +262,22 @@ class TestSymmetricTwoportAdaptor:
         )
 
 
+class TestReciprocal:
+    """Tests for Absolute class."""
+
+    def test_reciprocal_positive(self):
+        test_operation = Reciprocal()
+        assert test_operation.evaluate_output(0, [2]) == 0.5
+
+    def test_reciprocal_negative(self):
+        test_operation = Reciprocal()
+        assert test_operation.evaluate_output(0, [-5]) == -0.2
+
+    def test_reciprocal_complex(self):
+        test_operation = Reciprocal()
+        assert test_operation.evaluate_output(0, [1 + 1j]) == 0.5 - 0.5j
+
+
 class TestDepends:
     def test_depends_addition(self):
         add1 = Addition()

test/test_operation.py

 """
 B-ASIC test suite for the AbstractOperation class.
 """
+import re
+
+import pytest
 
 from b_asic import (
     MAD,
@@ -10,6 +13,7 @@ from b_asic import (
     ConstantMultiplication,
     Division,
     Multiplication,
+    Reciprocal,
     SquareRoot,
     Subtraction,
 )
@@ -100,6 +104,10 @@ class TestOperationOverloading:
         assert div3.input(0).signals[0].source.operation.value == 5
         assert div3.input(1).signals == div2.output(0).signals
 
+        div4 = 1 / div3
+        assert isinstance(div4, Reciprocal)
+        assert div4.input(0).signals == div3.output(0).signals
+
 
 class TestTraverse:
     def test_traverse_single_tree(self, operation):
@@ -212,18 +220,6 @@ class TestLatency:
             "out1": 9,
         }
 
-    def test_set_latency_offsets(self):
-        bfly = Butterfly()
-
-        bfly.set_latency_offsets({"in0": 3, "out1": 5})
-
-        assert bfly.latency_offsets == {
-            "in0": 3,
-            "in1": None,
-            "out0": None,
-            "out1": 5,
-        }
-
 
 class TestExecutionTime:
     def test_execution_time_constructor(self):
@@ -235,6 +231,13 @@ class TestExecutionTime:
 
         assert bfly.execution_time == 3
 
+    def test_set_execution_time_negative(self):
+        bfly = Butterfly()
+        with pytest.raises(
+            ValueError, match="Execution time cannot be negative"
+        ):
+            bfly.execution_time = -1
+
 
 class TestCopyOperation:
     def test_copy_butterfly_latency_offsets(self):
@@ -319,3 +322,47 @@ class TestSplit:
         assert len(split) == 2
         assert sum(isinstance(op, Addition) for op in split) == 1
         assert sum(isinstance(op, Subtraction) for op in split) == 1
+
+
+class TestLatencyOffset:
+    def test_set_latency_offsets(self):
+        bfly = Butterfly()
+
+        bfly.set_latency_offsets({"in0": 3, "out1": 5})
+
+        assert bfly.latency_offsets == {
+            "in0": 3,
+            "in1": None,
+            "out0": None,
+            "out1": 5,
+        }
+
+    def test_set_latency_offsets_error(self):
+        bfly = Butterfly()
+
+        with pytest.raises(
+            ValueError,
+            match=re.escape(
+                "Incorrectly formatted index in string, expected 'in' + index,"
+                " got: 'ina'"
+            ),
+        ):
+            bfly.set_latency_offsets({"ina": 3, "out1": 5})
+
+        with pytest.raises(
+            ValueError,
+            match=re.escape(
+                "Incorrectly formatted index in string, expected 'out' +"
+                " index, got: 'outb'"
+            ),
+        ):
+            bfly.set_latency_offsets({"in1": 3, "outb": 5})
+
+        with pytest.raises(
+            ValueError,
+            match=re.escape(
+                "Incorrectly formatted string, expected 'in' + index or 'out'"
+                " + index, got: 'foo'"
+            ),
+        ):
+            bfly.set_latency_offsets({"foo": 3, "out2": 5})

test/test_schedule.py

 """
 B-ASIC test suite for the schedule module and Schedule class.
 """
+import re
+
 import pytest
 
-from b_asic import Addition, ConstantMultiplication, Schedule
+from b_asic.core_operations import Addition, Butterfly, ConstantMultiplication
+from b_asic.schedule import Schedule
+from b_asic.signal_flow_graph import SFG
+from b_asic.special_operations import Input, Output
 
 
 class TestInit:
@@ -31,7 +36,7 @@ class TestInit:
             ConstantMultiplication.type_name(), 3
         )
 
-        schedule = Schedule(precedence_sfg_delays, scheduling_alg="ASAP")
+        schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
 
         for op in schedule._sfg.get_operations_topological_order():
             print(op.latency_offsets)
@@ -122,7 +127,7 @@ class TestInit:
             {"in0": 6, "in1": 7, "out0": 9}
         )
 
-        schedule = Schedule(precedence_sfg_delays, scheduling_alg="ASAP")
+        schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
 
         start_times_names = {}
         for op_id, start_time in schedule._start_times.items():
@@ -152,7 +157,7 @@ class TestInit:
     ):
         schedule = Schedule(
             sfg_two_inputs_two_outputs_independent_with_cmul,
-            scheduling_alg="ASAP",
+            scheduling_algorithm="ASAP",
         )
 
         start_times_names = {}
@@ -187,7 +192,7 @@ class TestSlacks:
             ConstantMultiplication.type_name(), 3
         )
 
-        schedule = Schedule(precedence_sfg_delays, scheduling_alg="ASAP")
+        schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
         assert (
             schedule.forward_slack(
                 precedence_sfg_delays.find_by_name("ADD3")[0].graph_id
@@ -220,7 +225,7 @@ class TestSlacks:
             ConstantMultiplication.type_name(), 3
         )
 
-        schedule = Schedule(precedence_sfg_delays, scheduling_alg="ASAP")
+        schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
         assert schedule.slacks(
             precedence_sfg_delays.find_by_name("ADD3")[0].graph_id
         ) == (0, 7)
@@ -236,7 +241,7 @@ class TestRescheduling:
             ConstantMultiplication.type_name(), 3
         )
 
-        schedule = Schedule(precedence_sfg_delays, scheduling_alg="ASAP")
+        schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
 
         schedule.move_operation(
             precedence_sfg_delays.find_by_name("ADD3")[0].graph_id, 4
@@ -274,7 +279,7 @@ class TestRescheduling:
             ConstantMultiplication.type_name(), 3
         )
 
-        schedule = Schedule(precedence_sfg_delays, scheduling_alg="ASAP")
+        schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
         add3_id = precedence_sfg_delays.find_by_name("ADD3")[0].graph_id
         schedule.move_operation(add3_id, 4)
         assert schedule.forward_slack(add3_id) == 3
@@ -300,7 +305,7 @@ class TestRescheduling:
             ConstantMultiplication.type_name(), 3
         )
 
-        schedule = Schedule(precedence_sfg_delays, scheduling_alg="ASAP")
+        schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
         with pytest.raises(ValueError):
             schedule.move_operation(
                 precedence_sfg_delays.find_by_name("ADD3")[0].graph_id, -4
@@ -314,7 +319,7 @@ class TestRescheduling:
             ConstantMultiplication.type_name(), 3
         )
 
-        schedule = Schedule(precedence_sfg_delays, scheduling_alg="ASAP")
+        schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
         with pytest.raises(ValueError):
             schedule.move_operation(
                 precedence_sfg_delays.find_by_name("ADD3")[0].graph_id, 10
@@ -327,7 +332,7 @@ class TestTimeResolution:
     ):
         schedule = Schedule(
             sfg_two_inputs_two_outputs_independent_with_cmul,
-            scheduling_alg="ASAP",
+            scheduling_algorithm="ASAP",
         )
         old_schedule_time = schedule.schedule_time
         assert schedule.get_possible_time_resolution_decrements() == [1]
@@ -363,7 +368,7 @@ class TestTimeResolution:
     ):
         schedule = Schedule(
             sfg_two_inputs_two_outputs_independent_with_cmul,
-            scheduling_alg="ASAP",
+            scheduling_algorithm="ASAP",
         )
         old_schedule_time = schedule.schedule_time
 
@@ -404,7 +409,7 @@ class TestTimeResolution:
     ):
         schedule = Schedule(
             sfg_two_inputs_two_outputs_independent_with_cmul,
-            scheduling_alg="ASAP",
+            scheduling_algorithm="ASAP",
         )
         old_schedule_time = schedule.schedule_time
         assert schedule.get_possible_time_resolution_decrements() == [1]
@@ -473,3 +478,72 @@ class TestFigureGeneration:
     @pytest.mark.mpl_image_compare(remove_text=True, style='mpl20')
     def test__get_figure_no_execution_times(self, secondorder_iir_schedule):
         return secondorder_iir_schedule._get_figure()
+
+
+class TestErrors:
+    def test_no_latency(self, sfg_simple_filter):
+        with pytest.raises(
+            ValueError,
+            match="Input port 0 of operation add1 has no latency-offset.",
+        ):
+            Schedule(sfg_simple_filter)
+
+    def test_no_output_latency(self):
+        in1 = Input()
+        in2 = Input()
+        bfly = Butterfly(
+            in1, in2, latency_offsets={"in0": 4, "in1": 2, "out0": 10}
+        )
+        out1 = Output(bfly.output(0))
+        out2 = Output(bfly.output(1))
+        sfg = SFG([in1, in2], [out1, out2])
+        with pytest.raises(
+            ValueError,
+            match="Output port 1 of operation bfly1 has no latency-offset.",
+        ):
+            Schedule(sfg)
+        in1 = Input()
+        in2 = Input()
+        bfly1 = Butterfly(
+            in1, in2, latency_offsets={"in0": 4, "in1": 2, "out1": 10}
+        )
+        bfly2 = Butterfly(
+            bfly1.output(0),
+            bfly1.output(1),
+            latency_offsets={"in0": 4, "in1": 2, "out0": 10, "out1": 8},
+        )
+        out1 = Output(bfly2.output(0))
+        out2 = Output(bfly2.output(1))
+        sfg = SFG([in1, in2], [out1, out2])
+        with pytest.raises(
+            ValueError,
+            match="Output port 0 of operation bfly1 has no latency-offset.",
+        ):
+            Schedule(sfg)
+
+    def test_too_short_schedule_time(self, sfg_simple_filter):
+        sfg_simple_filter.set_latency_of_type(Addition.type_name(), 5)
+        sfg_simple_filter.set_latency_of_type(
+            ConstantMultiplication.type_name(), 4
+        )
+        with pytest.raises(
+            ValueError, match="Too short schedule time. Minimum is 9."
+        ):
+            Schedule(sfg_simple_filter, schedule_time=3)
+
+        schedule = Schedule(sfg_simple_filter)
+        with pytest.raises(
+            ValueError,
+            match=re.escape("New schedule time (3) too short, minimum: 9."),
+        ):
+            schedule.set_schedule_time(3)
+
+    def test_incorrect_scheduling_algorithm(self, sfg_simple_filter):
+        sfg_simple_filter.set_latency_of_type(Addition.type_name(), 1)
+        sfg_simple_filter.set_latency_of_type(
+            ConstantMultiplication.type_name(), 2
+        )
+        with pytest.raises(
+            NotImplementedError, match="No algorithm with name: foo defined."
+        ):
+            Schedule(sfg_simple_filter, scheduling_algorithm="foo")

test/test_sfg.py

 import io
 import random
+import re
 import string
 import sys
 from os import path, remove
@@ -20,6 +21,7 @@ from b_asic.core_operations import (
     Multiplication,
     SquareRoot,
     Subtraction,
+    SymmetricTwoportAdaptor,
 )
 from b_asic.save_load_structure import python_to_sfg, sfg_to_python
 
@@ -327,27 +329,24 @@ class TestReplaceComponents:
         sfg = SFG(outputs=[Output(large_operation_tree)])
         component_id = "addd1"
 
-        try:
+        with pytest.raises(
+            ValueError, match="No operation matching the criteria found"
+        ):
             sfg = sfg.replace_component(
                 Multiplication(name="Multi"), graph_id=component_id
             )
-        except AssertionError:
-            assert True
-        else:
-            assert False
 
     def test_not_equal_input(self, large_operation_tree):
         sfg = SFG(outputs=[Output(large_operation_tree)])
         component_id = "c1"
 
-        try:
+        with pytest.raises(
+            TypeError,
+            match="The input count may not differ between the operations",
+        ):
             sfg = sfg.replace_component(
                 Multiplication(name="Multi"), graph_id=component_id
             )
-        except AssertionError:
-            assert True
-        else:
-            assert False
 
 
 class TestConstructSFG:
@@ -626,10 +625,16 @@ class TestFindComponentsWithTypeName:
 
 class TestGetPrecedenceList:
     def test_inputs_delays(self, precedence_sfg_delays):
+        # No cached precedence list
+        assert precedence_sfg_delays._precedence_list is None
+
         precedence_list = precedence_sfg_delays.get_precedence_list()
 
         assert len(precedence_list) == 7
 
+        # Cached precedence list
+        assert len(precedence_sfg_delays._precedence_list) == 7
+
         assert set(
             [
                 port.operation.key(port.index, port.operation.name)
@@ -679,6 +684,11 @@ class TestGetPrecedenceList:
             ]
         ) == {"ADD4"}
 
+        # Trigger cache
+        precedence_list = precedence_sfg_delays.get_precedence_list()
+
+        assert len(precedence_list) == 7
+
     def test_inputs_constants_delays_multiple_outputs(
         self, precedence_sfg_delays_and_constants
     ):
@@ -1413,3 +1423,139 @@ class TestSFGGraph:
     def test_show_sfg_invalid_engine(self, sfg_simple_filter):
         with pytest.raises(ValueError):
             sfg_simple_filter.show_sfg(engine="ppddff")
+
+
+class TestSFGErrors:
+    def test_dangling_output(self):
+        in1 = Input()
+        in2 = Input()
+        adaptor = SymmetricTwoportAdaptor(0.5, in1, in2)
+        out1 = Output(adaptor.output(0))
+        # No error, maybe should be?
+        _ = SFG([in1, in2], [out1])
+
+    def test_unconnected_input_port(self):
+        in1 = Input()
+        adaptor = SymmetricTwoportAdaptor(0.5, in1)
+        out1 = Output(adaptor.output(0))
+        with pytest.raises(ValueError, match="Unconnected input port in SFG"):
+            SFG([in1], [out1])
+
+    def test_unconnected_output(self):
+        in1 = Input()
+        in2 = Input()
+        adaptor = SymmetricTwoportAdaptor(0.5, in1, in2)
+        out1 = Output(adaptor.output(0))
+        out2 = Output()
+        # No error, should be
+        SFG([in1, in2], [out1, out2])
+
+    def test_unconnected_input(self):
+        in1 = Input()
+        in2 = Input()
+        adaptor = SymmetricTwoportAdaptor(0.5, in1)
+        out1 = Output(adaptor.output(0))
+        out2 = Output(adaptor.output(1))
+        # Correct error?
+        with pytest.raises(ValueError, match="Unconnected input port in SFG"):
+            SFG([in1, in2], [out1, out2])
+
+    def test_duplicate_input(self):
+        in1 = Input()
+        in2 = Input()
+        adaptor = SymmetricTwoportAdaptor(0.5, in1, in2)
+        out1 = Output(adaptor.output(0))
+        out2 = Output(adaptor.output(1))
+        with pytest.raises(ValueError, match="Duplicate input operation"):
+            SFG([in1, in1], [out1, out2])
+
+    def test_duplicate_output(self):
+        in1 = Input()
+        in2 = Input()
+        adaptor = SymmetricTwoportAdaptor(0.5, in1, in2)
+        out1 = Output(adaptor.output(0))
+        out2 = Output(adaptor.output(1))
+        with pytest.raises(ValueError, match="Duplicate output operation"):
+            SFG([in1, in2], [out1, out1])
+
+    def test_unconnected_input_signal(self):
+        in1 = Input()
+        in2 = Input()
+        adaptor = SymmetricTwoportAdaptor(0.5, in1, in2)
+        out1 = Output(adaptor.output(0))
+        out2 = Output(adaptor.output(1))
+        signal = Signal()
+        with pytest.raises(
+            ValueError, match="Input signal #0 is missing destination in SFG"
+        ):
+            SFG([in1, in2], [out1, out2], [signal])
+
+    def test_unconnected_output_signal(self):
+        in1 = Input()
+        in2 = Input()
+        adaptor = SymmetricTwoportAdaptor(0.5, in1, in2)
+        out1 = Output(adaptor.output(0))
+        out2 = Output(adaptor.output(1))
+        signal = Signal()
+        with pytest.raises(
+            ValueError, match="Output signal #0 is missing source in SFG"
+        ):
+            SFG([in1, in2], [out1, out2], output_signals=[signal])
+
+    def test_duplicate_input_signal(self):
+        in1 = Input()
+        signal = Signal()
+        adaptor = SymmetricTwoportAdaptor(0.5, in1, signal)
+        out1 = Output(adaptor.output(0))
+        out2 = Output(adaptor.output(1))
+        with pytest.raises(ValueError, match="Duplicate input signal"):
+            SFG([in1], [out1, out2], [signal, signal])
+
+    def test_duplicate_output_signal(self):
+        in1 = Input()
+        in2 = Input()
+        adaptor = SymmetricTwoportAdaptor(0.5, in1, in2)
+        out1 = Output(adaptor.output(0))
+        signal = Signal(adaptor.output(1))
+        # Should raise?
+        SFG([in1, in2], [out1], output_signals=[signal, signal])
+
+    def test_dangling_input_signal(self):
+        in1 = Input()
+        signal = Signal()
+        adaptor = SymmetricTwoportAdaptor(0.5, in1, signal)
+        out1 = Output(adaptor.output(0))
+        out2 = Output(adaptor.output(1))
+        with pytest.raises(
+            ValueError, match="Dangling signal without source in SFG"
+        ):
+            SFG([in1], [out1, out2])
+
+    def test_remove_signal_with_different_number_of_inputs_and_outputs(self):
+        in1 = Input()
+        in2 = Input()
+        add1 = Addition(in1, in2, name="addition")
+        out1 = Output(add1)
+        sfg = SFG([in1, in2], [out1])
+        # Try to remove non-existent operation
+        sfg1 = sfg.remove_operation("foo")
+        assert sfg1 is None
+        with pytest.raises(
+            ValueError,
+            match=(
+                "Different number of input and output ports of operation with"
+            ),
+        ):
+            sfg.remove_operation('add1')
+
+    def test_inputs_required_for_output(self):
+        in1 = Input()
+        in2 = Input()
+        add1 = Addition(in1, in2, name="addition")
+        out1 = Output(add1)
+        sfg = SFG([in1, in2], [out1])
+        with pytest.raises(
+            IndexError,
+            match=re.escape("Output index out of range (expected 0-0, got 1)"),
+        ):
+            sfg.inputs_required_for_output(1)

test/test_sfg_generator.py

+from b_asic.core_operations import SymmetricTwoportAdaptor
+from b_asic.sfg_generator import wdf_allpass
+
+
+def test_wdf_allpass():
+    sfg = wdf_allpass([0.3, 0.5, 0.7])
+    assert (
+        len(
+            [
+                comp
+                for comp in sfg.components
+                if isinstance(comp, SymmetricTwoportAdaptor)
+            ]
+        )
+        == 3
+    )
+
+    sfg = wdf_allpass([0.3, 0.5, 0.7, 0.9])
+    assert (
+        len(
+            [
+                comp
+                for comp in sfg.components
+                if isinstance(comp, SymmetricTwoportAdaptor)
+            ]
+        )
+        == 4
+    )

test/test_signal_generator.py

+from math import sqrt
+
+import pytest
+
+from b_asic.signal_generator import Constant, Impulse, Sinusoid, Step, ZeroPad
+
+
+def test_impulse():
+    g = Impulse()
+    assert g(-1) == 0
+    assert g(0) == 1
+    assert g(1) == 0
+    assert g(2) == 0
+
+    g = Impulse(1)
+    assert g(-1) == 0
+    assert g(0) == 0
+    assert g(1) == 1
+    assert g(2) == 0
+
+
+def test_step():
+    g = Step()
+    assert g(-1) == 0
+    assert g(0) == 1
+    assert g(1) == 1
+    assert g(2) == 1
+
+    g = Step(1)
+    assert g(-1) == 0
+    assert g(0) == 0
+    assert g(1) == 1
+    assert g(2) == 1
+
+
+def test_constant():
+    g = Constant()
+    assert g(-1) == 1
+    assert g(0) == 1
+    assert g(1) == 1
+    assert g(2) == 1
+
+    g = Constant(0.5)
+    assert g(-1) == 0.5
+    assert g(0) == 0.5
+    assert g(1) == 0.5
+    assert g(2) == 0.5
+
+
+def test_zeropad():
+    g = ZeroPad([0.4, 0.6])
+    assert g(-1) == 0
+    assert g(0) == 0.4
+    assert g(1) == 0.6
+    assert g(2) == 0
+
+
+def test_sinusoid():
+    g = Sinusoid(0.5)
+    assert g(0) == 0
+    assert g(1) == 1
+    assert g(2) == pytest.approx(0)
+    assert g(3) == -1
+
+    g = Sinusoid(0.5, 0.25)
+    assert g(0) == pytest.approx(sqrt(2) / 2)
+    assert g(1) == pytest.approx(sqrt(2) / 2)
+    assert g(2) == pytest.approx(-sqrt(2) / 2)
+    assert g(3) == pytest.approx(-sqrt(2) / 2)
+
+
+def test_addition():
+    g = Impulse() + Impulse(2)
+    assert g(-1) == 0
+    assert g(0) == 1
+    assert g(1) == 0
+    assert g(2) == 1
+    assert g(3) == 0
+
+    g = 1 + Impulse(2)
+    assert g(-1) == 1
+    assert g(0) == 1
+    assert g(1) == 1
+    assert g(2) == 2
+    assert g(3) == 1
+
+    g = Impulse(1) + 1
+    assert g(-1) == 1
+    assert g(0) == 1
+    assert g(1) == 2
+    assert g(2) == 1
+    assert g(3) == 1
+
+
+def test_subtraction():
+    g = Impulse() - Impulse(2)
+    assert g(-1) == 0
+    assert g(0) == 1
+    assert g(1) == 0
+    assert g(2) == -1
+    assert g(3) == 0
+
+    g = 1 - Impulse(2)
+    assert g(-1) == 1
+    assert g(0) == 1
+    assert g(1) == 1
+    assert g(2) == 0
+    assert g(3) == 1
+
+    g = Impulse(2) - 1
+    assert g(-1) == -1
+    assert g(0) == -1
+    assert g(1) == -1
+    assert g(2) == 0
+    assert g(3) == -1
+
+
+def test_multiplication():
+    g = Impulse() * 0.5
+    assert g(-1) == 0
+    assert g(0) == 0.5
+    assert g(1) == 0
+    assert g(2) == 0
+
+    g = 2 * Sinusoid(0.5, 0.25)
+    assert g(0) == pytest.approx(sqrt(2))
+    assert g(1) == pytest.approx(sqrt(2))
+    assert g(2) == pytest.approx(-sqrt(2))
+    assert g(3) == pytest.approx(-sqrt(2))