add some more advanced sfg simulation test cases

b_asic/simulation.py

@@ -5,7 +5,7 @@ TODO: More info.
 
 from collections import defaultdict
 from numbers import Number
-from typing import List, Dict, DefaultDict, Callable, Sequence, Mapping
+from typing import List, Dict, DefaultDict, Callable, Sequence, Mapping, Union, Optional
 
 from b_asic.signal_flow_graph import SFG
 
@@ -23,28 +23,34 @@ class Simulation:
     _current_input_values: Sequence[Number]
     _latest_output_values: Sequence[Number]
 
-    def __init__(self, sfg: SFG, input_functions: Sequence[Callable[[int], Number]]):
-        if len(input_functions) != sfg.input_count:
-            raise ValueError(f"Wrong number of inputs supplied to simulation (expected {len(sfg.input_count)}, got {len(input_functions)})")
+    def __init__(self, sfg: SFG, input_providers: Optional[Sequence[Union[Sequence[Number], Callable[[int], Number]]]] = None):
         self._sfg = sfg
         self._results = defaultdict(dict)
         self._registers = {}
         self._iteration = 0
-        self._input_functions = list(input_functions)
-        self._current_input_values = [0 for _ in range(self._sfg.input_count)]
+        self._input_functions = []
+        self._current_input_values = []
         self._latest_output_values = [0 for _ in range(self._sfg.output_count)]
+        if input_providers is not None:
+            self.set_inputs(input_providers)
 
-    def input_functions(self, input_functions: Sequence[Callable[[int], Number]]) -> None:
+    def set_inputs(self, input_providers: Sequence[Union[Sequence[Number], Callable[[int], Number]]]) -> None:
         """Set the input functions used to get values for the inputs to the internal SFG."""
-        if len(input_functions) != len(self._input_functions):
-            raise ValueError(f"Wrong number of inputs supplied to simulation (expected {len(self._input_functions)}, got {len(input_functions)})")
-        self._input_functions = input_functions
+        if len(input_providers) != self._sfg.input_count:
+            raise ValueError(f"Wrong number of inputs supplied to simulation (expected {self._sfg.input_count}, got {len(input_providers)})")
+        self._input_functions = [None for _ in range(self._sfg.input_count)]
+        for index, input_provider in enumerate(input_providers):
+            self.set_input(index, input_provider)
 
-    def input_function(self, index: int, input_function: Callable[[int], Number]) -> None:
+    def set_input(self, index: int, input_provider: Union[Sequence[Number], Callable[[int], Number]]) -> None:
         """Set the input function used to get values for the specific input at the given index to the internal SFG."""
         if index < 0 or index >= len(self._input_functions):
             raise IndexError(f"Input index out of range (expected 0-{self._input_functions - 1}, got {index})")
-        self._input_functions[index] = input_function
+
+        if callable(input_provider):
+            self._input_functions[index] = input_provider
+        else:
+            self._input_functions[index] = lambda n: input_provider[n]
 
     def run(self) -> Sequence[Number]:
         """Run one iteration of the simulation and return the resulting output values."""

test/fixtures/signal_flow_graph.py

 import pytest
 
-from b_asic import SFG, Input, Output, Addition
+from b_asic import SFG, Input, Output, Constant, Register
 
 
 @pytest.fixture
 def sfg_two_inputs_two_outputs():
-    """Valid SFG containing two inputs and two outputs.
+    """Valid SFG with two inputs and two outputs.
 	     .                           .
     in1>------+        +---------------out1>
          .    |        |             .
@@ -19,12 +19,42 @@ def sfg_two_inputs_two_outputs():
     out1 = in1 + in2
 	out2 = in1 + 2 * in2
     """
-    in1=Input()
-    in2=Input()
-    add1=Addition(in1, in2)
-    add2=Addition(add1, in2)
-    out1=Output(add1)
-    out2=Output(add2)
+    in1 = Input()
+    in2 = Input()
+    add1 = in1 + in2
+    add2 = add1 + in2
+    out1 = Output(add1)
+    out2 = Output(add2)
     return SFG(inputs = [in1, in2], outputs = [out1, out2])
 
-# TODO: Testa nestad sfg
\ No newline at end of file
+@pytest.fixture
+def sfg_nested():
+    """Valid SFG with two inputs and one output.
+    out1 = in1 + (in1 + in1 * in2) * (in1 + in2 * (in1 + in1 * in2))
+    """
+    mac_in1 = Input()
+    mac_in2 = Input()
+    mac_in3 = Input()
+    mac_out1 = Output(mac_in1 + mac_in2 * mac_in3)
+    MAC = SFG(inputs = [mac_in1, mac_in2, mac_in3], outputs=[mac_out1])
+
+    in1 = Input()
+    in2 = Input()
+    mac1 = MAC(in1, in1, in2)
+    mac2 = MAC(in1, in2, mac1)
+    mac3 = MAC(in1, mac1, mac2)
+    out1 = Output(mac3)
+    return SFG(inputs = [in1, in2], outputs=[out1])
+
+@pytest.fixture
+def sfg_accumulator():
+    """Valid SFG with two inputs and one output.
+    data_out = (data_in' + data_in) * (1 - reset)
+    """
+    data_in = Input()
+    reset = Input()
+    reset_inverted = Constant(1) - reset
+    reg = Register()
+    reg.input(0).connect((reg + data_in) * reset_inverted)
+    data_out = Output(reg)
+    return SFG(inputs = [data_in, reset], outputs = [data_out])
\ No newline at end of file

test/test_simulation.py

+import pytest
+import numpy as np
+
 from b_asic import SFG, Output, Simulation
 
 
 class TestSimulation:
-    def test_simulate(self, sfg_two_inputs_two_outputs):
+    def test_simulate_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])
+
         output = simulation.run_for(101)
+
         assert output[0] == 304
         assert output[1] == 505
 
+        assert simulation.results[100]["0"] == 304
+        assert simulation.results[100]["1"] == 505
+
         assert simulation.results[0]["in1"] == 3
         assert simulation.results[0]["in2"] == 1
         assert simulation.results[0]["add1"] == 4
@@ -35,3 +43,79 @@ class TestSimulation:
         assert simulation.results[3]["add2"] == 20
         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):
+        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])
+
+        output = simulation.run_for(5)
+
+        assert output[0] == 9
+        assert output[1] == 11
+
+        assert simulation.results[4]["0"] == 9
+        assert simulation.results[4]["1"] == 11
+
+        assert simulation.results[0]["in1"] == 5
+        assert simulation.results[0]["in2"] == 7
+        assert simulation.results[0]["add1"] == 12
+        assert simulation.results[0]["add2"] == 19
+        assert simulation.results[0]["0"] == 12
+        assert simulation.results[0]["1"] == 19
+
+        assert simulation.results[1]["in1"] == 9
+        assert simulation.results[1]["in2"] == 3
+        assert simulation.results[1]["add1"] == 12
+        assert simulation.results[1]["add2"] == 15
+        assert simulation.results[1]["0"] == 12
+        assert simulation.results[1]["1"] == 15
+
+        assert simulation.results[2]["in1"] == 25
+        assert simulation.results[2]["in2"] == 3
+        assert simulation.results[2]["add1"] == 28
+        assert simulation.results[2]["add2"] == 31
+        assert simulation.results[2]["0"] == 28
+        assert simulation.results[2]["1"] == 31
+
+        assert simulation.results[3]["in1"] == -5
+        assert simulation.results[3]["in2"] == 54
+        assert simulation.results[3]["add1"] == 49
+        assert simulation.results[3]["add2"] == 103
+        assert simulation.results[3]["0"] == 49
+        assert simulation.results[3]["1"] == 103
+
+    def test_simulate_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])
+        simulation.run_for(5)
+        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])
+
+        output1 = simulation.run()
+        output2 = simulation.run()
+
+        assert output1[0] == 11405
+        assert output2[0] == 8109
+
+    def test_simulate_with_register(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])
+
+        output = simulation.run_for(6)
+
+        assert output[0] == 7
+
+        assert simulation.results[0]["0"] == 0
+        assert simulation.results[1]["0"] == 5
+        assert simulation.results[2]["0"] == 3
+        assert simulation.results[3]["0"] == 28
+        assert simulation.results[4]["0"] == 0
+        assert simulation.results[5]["0"] == 7
\ No newline at end of file