Refactor constructor so that Input signals and Output signals are connected to...

Refactor constructor so that Input signals and Output signals are connected to ports before traversal is started, that way edge cases of empty SFG's are easily handled

test/fixtures/port.py

@@ -10,3 +10,11 @@ def input_port():
 @pytest.fixture
 def output_port():
     return OutputPort(None, 0)
+
+@pytest.fixture
+def list_of_input_ports():
+    return [InputPort(None, i) for i in range(0, 3)]
+
+@pytest.fixture
+def list_of_output_ports():
+    return [OutputPort(None, i) for i in range(0, 3)]

test/fixtures/signal_flow_graph.py

 import pytest
 
-from b_asic import SFG, Input, Output, Constant, Register
+from b_asic import SFG, Input, Output, Constant, Register, ConstantMultiplication
 
 
 @pytest.fixture
@@ -95,4 +95,21 @@ def sfg_accumulator():
     reg = Register()
     reg.input(0).connect((reg + data_in) * (1 - reset))
     data_out = Output(reg)
-    return SFG(inputs = [data_in, reset], outputs = [data_out])
\ No newline at end of file
+    return SFG(inputs = [data_in, reset], outputs = [data_out])
+
+@pytest.fixture
+def simple_filter():
+    """A valid SFG that is used as a filter in the first lab for TSTE87.
+                +----<constmul1----+
+                |                  |
+                |                  |
+    in1>------add1>------reg>------+------out1>
+    """
+    in1 = Input()
+    reg = Register()
+    constmul1 = ConstantMultiplication(0.5)
+    add1 = in1 + constmul1
+    reg.input(0).connect(add1)
+    constmul1.input(0).connect(reg)
+    out1 = Output(reg)
+    return SFG(inputs=[in1], outputs=[out1])

test/test_inputport.py

@@ -7,15 +7,7 @@ import pytest
 from b_asic import InputPort, OutputPort, Signal
 
 @pytest.fixture
-def inp_port():
-    return InputPort(None, 0)
-
-@pytest.fixture
-def out_port():
-    return OutputPort(None, 0)
-
-@pytest.fixture
-def out_port2():
+def output_port2():
     return OutputPort(None, 1)
 
 @pytest.fixture
@@ -23,53 +15,53 @@ def dangling_sig():
     return Signal()
 
 @pytest.fixture
-def s_w_source(out_port):
-    return Signal(source=out_port)
+def s_w_source(output_port):
+    return Signal(source=output_port)
 
 @pytest.fixture
 def sig_with_dest(inp_port):
     return Signal(destination=inp_port)
 
 @pytest.fixture
-def connected_sig(inp_port, out_port):
-    return Signal(source=out_port, destination=inp_port)
+def connected_sig(inp_port, output_port):
+    return Signal(source=output_port, destination=inp_port)
 
-def test_connect_then_disconnect(inp_port, out_port):
+def test_connect_then_disconnect(input_port, output_port):
     """Test connect unused port to port."""
-    s1 = inp_port.connect(out_port)
+    s1 = input_port.connect(output_port)
 
-    assert inp_port.connected_source == out_port
-    assert inp_port.signals == [s1]
-    assert out_port.signals == [s1]
-    assert s1.source is out_port
-    assert s1.destination is inp_port
+    assert input_port.connected_source == output_port
+    assert input_port.signals == [s1]
+    assert output_port.signals == [s1]
+    assert s1.source is output_port
+    assert s1.destination is input_port
 
-    inp_port.remove_signal(s1)
+    input_port.remove_signal(s1)
 
-    assert inp_port.connected_source is None
-    assert inp_port.signals == []
-    assert out_port.signals == [s1]
-    assert s1.source is out_port
+    assert input_port.connected_source is None
+    assert input_port.signals == []
+    assert output_port.signals == [s1]
+    assert s1.source is output_port
     assert s1.destination is None
 
-def test_connect_used_port_to_new_port(inp_port, out_port, out_port2):
-    """Does connecting multiple ports to an inputport throw error?"""
-    inp_port.connect(out_port)
+def test_connect_used_port_to_new_port(input_port, output_port, output_port2):
+    """Multiple connections to an input port should throw an error."""
+    input_port.connect(output_port)
     with pytest.raises(Exception):
-        inp_port.connect(out_port2)
+        input_port.connect(output_port2)
 
-def test_add_signal_then_disconnect(inp_port, s_w_source):
+def test_add_signal_then_disconnect(input_port, s_w_source):
     """Can signal be connected then disconnected properly?"""
-    inp_port.add_signal(s_w_source)
+    input_port.add_signal(s_w_source)
 
-    assert inp_port.connected_source == s_w_source.source
-    assert inp_port.signals == [s_w_source]
+    assert input_port.connected_source == s_w_source.source
+    assert input_port.signals == [s_w_source]
     assert s_w_source.source.signals == [s_w_source]
-    assert s_w_source.destination is inp_port
+    assert s_w_source.destination is input_port
 
-    inp_port.remove_signal(s_w_source)
+    input_port.remove_signal(s_w_source)
 
-    assert inp_port.connected_source is None
-    assert inp_port.signals == []
+    assert input_port.connected_source is None
+    assert input_port.signals == []
     assert s_w_source.source.signals == [s_w_source]
     assert s_w_source.destination is None

test/test_outputport.py

@@ -5,33 +5,19 @@ import pytest
 
 from b_asic import OutputPort, InputPort, Signal
 
-
-@pytest.fixture
-def output_port():
-    return OutputPort(None, 0)
-
-@pytest.fixture
-def input_port():
-    return InputPort(None, 0)
-
-@pytest.fixture
-def list_of_input_ports():
-    return [InputPort(None, i) for i in range(0, 3)]
-
-
 class TestConnect:
     def test_multiple_ports(self, output_port, list_of_input_ports):
-        """Can multiple ports connect to an output port?"""
+        """Multiple connections to an output port should be possible."""
         for port in list_of_input_ports:
             port.connect(output_port)
 
         assert output_port.signal_count == len(list_of_input_ports)
 
-    def test_same_port(self, output_port, list_of_input_ports):
+    def test_same_port(self, output_port, input_port):
         """Check error handing."""
-        list_of_input_ports[0].connect(output_port)
+        input_port.connect(output_port)
         with pytest.raises(Exception):
-            list_of_input_ports[0].connect(output_port)
+            input_port.connect(output_port)
 
         assert output_port.signal_count == 1
 
@@ -43,9 +29,8 @@ class TestAddSignal:
         assert output_port.signal_count == 1
         assert output_port.signals == [s]
 
-class TestDisconnect:
+class TestClear:
     def test_others_clear(self, output_port, list_of_input_ports):
-        """Can multiple ports disconnect from OutputPort?"""
         for port in list_of_input_ports:
             port.connect(output_port)
 
@@ -56,7 +41,6 @@ class TestDisconnect:
         assert all(s.dangling() for s in output_port.signals)
 
     def test_self_clear(self, output_port, list_of_input_ports):
-        """Can an OutputPort disconnect from multiple ports?"""
         for port in list_of_input_ports:
             port.connect(output_port)
 
@@ -74,7 +58,6 @@ class TestRemoveSignal:
         assert output_port.signals == []
 
     def test_multiple_signals(self, output_port, list_of_input_ports):
-        """Can multiple signals disconnect from OutputPort?"""
         sigs = []
 
         for port in list_of_input_ports:

test/test_print_sfg.py

-"""
-B-ASIC test suite for printing a SFG
-"""
-
-
-from b_asic.signal_flow_graph import SFG
-from b_asic.core_operations import Addition, Multiplication, Constant
-from b_asic.port import InputPort, OutputPort
-from b_asic.signal import Signal
-from b_asic.special_operations import Input, Output
-
-import pytest
-
-
-class TestPrintSfg:
-    def test_print_one_addition(self):
-        inp1 = Input("INP1")
-        inp2 = Input("INP2")
-        add1 = Addition(inp1, inp2, "ADD1")
-        out1 = Output(add1, "OUT1")
-        sfg = SFG(inputs=[inp1, inp2], outputs=[out1], name="sf1")
-
-        assert sfg.__str__() == ("id: add1, name: ADD1, input: [s1, s2], output: [s3]\nid: in1, name: INP1, input: [], output: [s1]\nid: in2, name: INP2, input: [], output: [s2]\nid: out1, name: OUT1, input: [s3], output: []\n")
-
-    def test_print_add_mul(self):
-        inp1 = Input("INP1")
-        inp2 = Input("INP2")
-        inp3 = Input("INP3")
-        add1 = Addition(inp1, inp2, "ADD1")
-        mul1 = Multiplication(add1, inp3, "MUL1")
-        out1 = Output(mul1, "OUT1")
-        sfg = SFG(inputs=[inp1, inp2, inp3], outputs=[out1], name="mac_sfg")
-
-        assert sfg.__str__() == ("id: add1, name: ADD1, input: [s1, s2], output: [s5]\nid: in1, name: INP1, input: [], output: [s1]\nid: in2, name: INP2, input: [], output: [s2]\nid: mul1, name: MUL1, input: [s5, s3], output: [s4]\nid: in3, name: INP3, input: [], output: [s3]\nid: out1, name: OUT1, input: [s4], output: []\n")
-
-    def test_print_constant(self):
-        inp1 = Input("INP1")
-        const1 = Constant(3, "CONST")
-        add1 = Addition(const1, inp1, "ADD1")
-        out1 = Output(add1, "OUT1")
-
-        sfg = SFG(inputs=[inp1], outputs=[out1], name="sfg")
-
-        assert sfg.__str__() == ("id: add1, name: ADD1, input: [s3, s1], output: [s2]\nid: c1, name: CONST, value: 3, input: [], output: [s3]\nid: in1, name: INP1, input: [], output: [s1]\nid: out1, name: OUT1, input: [s2], output: []\n")
-
-     
\ No newline at end of file

test/test_signal_flow_graph.py → test/test_sfg.py

@@ -3,7 +3,7 @@ import pytest
 from b_asic import SFG, Signal, Input, Output, Constant, Addition, Multiplication
 
 
-class TestConstructor:
+class TestInit:
     def test_direct_input_to_output_sfg_construction(self):
         in1 = Input("IN1")
         out1 = Output(None, "OUT1")
@@ -45,21 +45,58 @@ class TestConstructor:
         assert sfg.input_count == 0
         assert sfg.output_count == 1
 
+class TestPrintSfg:
+    def test_one_addition(self):
+        inp1 = Input("INP1")
+        inp2 = Input("INP2")
+        add1 = Addition(inp1, inp2, "ADD1")
+        out1 = Output(add1, "OUT1")
+        sfg = SFG(inputs=[inp1, inp2], outputs=[out1], name="sf1")
 
-class TestEvaluation:
-    def test_evaluate_output(self, operation_tree):
-        sfg = SFG(outputs = [Output(operation_tree)])
-        assert sfg.evaluate_output(0, []) == 5
+        assert sfg.__str__() == \
+            "id: add1, name: ADD1, input: [s1, s2], output: [s3]\n" + \
+            "id: in1, name: INP1, input: [], output: [s1]\n" + \
+            "id: in2, name: INP2, input: [], output: [s2]\n" + \
+            "id: out1, name: OUT1, input: [s3], output: []\n"
 
-    def test_evaluate_output_large(self, large_operation_tree):
-        sfg = SFG(outputs = [Output(large_operation_tree)])
-        assert sfg.evaluate_output(0, []) == 14
+    def test_add_mul(self):
+        inp1 = Input("INP1")
+        inp2 = Input("INP2")
+        inp3 = Input("INP3")
+        add1 = Addition(inp1, inp2, "ADD1")
+        mul1 = Multiplication(add1, inp3, "MUL1")
+        out1 = Output(mul1, "OUT1")
+        sfg = SFG(inputs=[inp1, inp2, inp3], outputs=[out1], name="mac_sfg")
 
-    def test_evaluate_output_cycle(self, operation_graph_with_cycle):
-        sfg = SFG(outputs = [Output(operation_graph_with_cycle)])
-        with pytest.raises(Exception):
-            sfg.evaluate_output(0, [])
+        assert sfg.__str__() == \
+            "id: add1, name: ADD1, input: [s1, s2], output: [s5]\n" + \
+            "id: in1, name: INP1, input: [], output: [s1]\n" + \
+            "id: in2, name: INP2, input: [], output: [s2]\n" + \
+            "id: mul1, name: MUL1, input: [s5, s3], output: [s4]\n" + \
+            "id: in3, name: INP3, input: [], output: [s3]\n" + \
+            "id: out1, name: OUT1, input: [s4], output: []\n"
 
+    def test_constant(self):
+        inp1 = Input("INP1")
+        const1 = Constant(3, "CONST")
+        add1 = Addition(const1, inp1, "ADD1")
+        out1 = Output(add1, "OUT1")
+
+        sfg = SFG(inputs=[inp1], outputs=[out1], name="sfg")
+
+        assert sfg.__str__() == \
+            "id: add1, name: ADD1, input: [s3, s1], output: [s2]\n" + \
+            "id: c1, name: CONST, value: 3, input: [], output: [s3]\n" + \
+            "id: in1, name: INP1, input: [], output: [s1]\n" + \
+            "id: out1, name: OUT1, input: [s2], output: []\n"
+
+    def test_simple_filter(self, simple_filter):
+         assert simple_filter.__str__() == \
+            'id: add1, name: , input: [s1, s3], output: [s4]\n' + \
+            'id: in1, name: , input: [], output: [s1]\n' + \
+            'id: cmul1, name: , input: [s5], output: [s3]\n' + \
+            'id: reg1, name: , input: [s4], output: [s5, s2]\n' + \
+            'id: out1, name: , input: [s2], output: []\n'
 
 class TestDeepCopy:
     def test_deep_copy_no_duplicates(self):
@@ -125,6 +162,19 @@ class TestDeepCopy:
         assert mac_sfg_new.input(0).signals[0].source.operation is a
         assert mac_sfg_new.input(1).signals[0].source.operation is b
 
+class TestEvaluateOutput:
+    def test_evaluate_output(self, operation_tree):
+        sfg = SFG(outputs = [Output(operation_tree)])
+        assert sfg.evaluate_output(0, []) == 5
+
+    def test_evaluate_output_large(self, large_operation_tree):
+        sfg = SFG(outputs = [Output(large_operation_tree)])
+        assert sfg.evaluate_output(0, []) == 14
+
+    def test_evaluate_output_cycle(self, operation_graph_with_cycle):
+        sfg = SFG(outputs = [Output(operation_graph_with_cycle)])
+        with pytest.raises(Exception):
+            sfg.evaluate_output(0, [])
 
 class TestComponents:
     def test_advanced_components(self):
@@ -150,7 +200,6 @@ class TestComponents:
 
 
 class TestReplaceComponents:
-
     def test_replace_addition_by_id(self, operation_tree):
         sfg = SFG(outputs=[Output(operation_tree)])
         component_id = "add1"

test/test_signal.py

@@ -61,27 +61,28 @@ def test_signal_creation_and_disconnction_and_connection_changing():
     assert s.source is out_port
     assert s.destination is in_port
 
-def test_signal_set_bits_pos_int(signal):
-    signal.bits = 10
-    assert signal.bits == 10
-
-def test_signal_set_bits_zero(signal):
-    signal.bits = 0
-    assert signal.bits == 0
-
-def test_signal_set_bits_neg_int(signal):
-    with pytest.raises(Exception):
-        signal.bits = -10
-
-def test_signal_set_bits_complex(signal):
-    with pytest.raises(Exception):
-        signal.bits = (2+4j)
-
-def test_signal_set_bits_float(signal):
-    with pytest.raises(Exception):
-        signal.bits = 3.2
-
-def test_signal_set_bits_pos_then_none(signal):
-    signal.bits = 10
-    signal.bits = None
-    assert signal.bits is None
\ No newline at end of file
+class Bits:
+    def test_pos_int(self, signal):
+        signal.bits = 10
+        assert signal.bits == 10
+
+    def test_bits_zero(self, signal):
+        signal.bits = 0
+        assert signal.bits == 0
+
+    def test_bits_neg_int(self, signal):
+        with pytest.raises(Exception):
+            signal.bits = -10
+
+    def test_bits_complex(self, signal):
+        with pytest.raises(Exception):
+            signal.bits = (2+4j)
+
+    def test_bits_float(self, signal):
+        with pytest.raises(Exception):
+            signal.bits = 3.2
+
+    def test_bits_pos_then_none(self, signal):
+        signal.bits = 10
+        signal.bits = None
+        assert signal.bits is None
\ No newline at end of file

test/test_simulation.py

@@ -4,8 +4,8 @@ import numpy as np
 from b_asic import SFG, Output, Simulation
 
 
-class TestSimulation:
-    def test_simulate_with_lambdas_as_input(self, sfg_two_inputs_two_outputs):
+class TestRunFor:
+    def test_with_lambdas_as_input(self, sfg_two_inputs_two_outputs):
         simulation = Simulation(sfg_two_inputs_two_outputs, [lambda n: n + 3, lambda n: 1 + n * 2], save_results = True)
 
         output = simulation.run_for(101)
@@ -44,7 +44,7 @@ class TestSimulation:
         assert simulation.results[3]["0"] == 13
         assert simulation.results[3]["1"] == 20
 
-    def test_simulate_with_numpy_arrays_as_input(self, sfg_two_inputs_two_outputs):
+    def test_with_numpy_arrays_as_input(self, sfg_two_inputs_two_outputs):
         input0 = np.array([5, 9, 25, -5, 7])
         input1 = np.array([7, 3, 3,  54, 2])
         simulation = Simulation(sfg_two_inputs_two_outputs, [input0, input1])
@@ -85,8 +85,8 @@ class TestSimulation:
 
         assert simulation.results[4]["0"] == 9
         assert simulation.results[4]["1"] == 11
-
-    def test_simulate_with_numpy_array_overflow(self, sfg_two_inputs_two_outputs):
+    
+    def test_with_numpy_array_overflow(self, sfg_two_inputs_two_outputs):
         input0 = np.array([5, 9, 25, -5, 7])
         input1 = np.array([7, 3, 3,  54, 2])
         simulation = Simulation(sfg_two_inputs_two_outputs, [input0, input1])
@@ -94,18 +94,7 @@ class TestSimulation:
         with pytest.raises(IndexError):
             simulation.run_for(1)
 
-    def test_simulate_nested(self, sfg_nested):
-        input0 = np.array([5, 9])
-        input1 = np.array([7, 3])
-        simulation = Simulation(sfg_nested, [input0, input1])
-
-        output0 = simulation.run()
-        output1 = simulation.run()
-
-        assert output0[0] == 11405
-        assert output1[0] == 4221
-
-    def test_simulate_delay(self, sfg_delay):
+    def test_delay(self, sfg_delay):
         simulation = Simulation(sfg_delay, save_results = True)
         simulation.set_input(0, [5, -2, 25, -6, 7, 0])
         simulation.run_for(6)
@@ -116,8 +105,20 @@ class TestSimulation:
         assert simulation.results[3]["0"] == 25
         assert simulation.results[4]["0"] == -6
         assert simulation.results[5]["0"] == 7
+
+class TestRun:
+    def test_nested(self, sfg_nested):
+        input0 = np.array([5, 9])
+        input1 = np.array([7, 3])
+        simulation = Simulation(sfg_nested, [input0, input1])
+
+        output0 = simulation.run()
+        output1 = simulation.run()
+
+        assert output0[0] == 11405
+        assert output1[0] == 4221
     
-    def test_simulate_accumulator(self, sfg_accumulator):
+    def test_accumulator(self, sfg_accumulator):
         data_in = np.array([5, -2, 25, -6, 7, 0])
         reset   = np.array([0, 0,  0,  1,  0, 0])
         simulation = Simulation(sfg_accumulator, [data_in, reset])
@@ -132,4 +133,10 @@ class TestSimulation:
         assert output2[0] == 3
         assert output3[0] == 28
         assert output4[0] == 0
-        assert output5[0] == 7
\ No newline at end of file
+        assert output5[0] == 7
+        
+    def test_simple_filter(self, simple_filter):
+        input0 = np.array([1, 2, 3, 4, 5])
+        simulation = Simulation(simple_filter, [input0], save_results=True)
+        output0 = [simulation.run()[0] for _ in range(len(input0))]
+        assert output0 == [0, 1.0, 2.5, 4.25, 6.125]