b_asic/simulation.py

@@ -3,33 +3,115 @@ B-ASIC Simulation Module.
 TODO: More info.
 """
 
+import numpy as np
+
+from collections import defaultdict
 from numbers import Number
-from typing import List
+from typing import List, Dict, DefaultDict, Callable, Sequence, Mapping, Union, Optional, MutableSequence, MutableMapping
+
+from b_asic.operation import ResultKey, ResultMap, MutableResultMap, MutableDelayMap
+from b_asic.signal_flow_graph import SFG
+
 
+ResultArrayMap = Mapping[ResultKey, Sequence[Number]]
+MutableResultArrayMap = MutableMapping[ResultKey, MutableSequence[Number]]
+InputFunction = Callable[[int], Number]
+InputProvider = Union[Number, Sequence[Number], InputFunction]
 
-class OperationState:
-    """Simulation state of an operation.
+
+class Simulation:
+    """Simulation.
     TODO: More info.
     """
 
-    output_values: List[Number]
-    iteration: int
+    _sfg: SFG
+    _results: MutableResultArrayMap
+    _delays: MutableDelayMap
+    _iteration: int
+    _input_functions: Sequence[InputFunction]
+    _input_length: Optional[int]
 
-    def __init__(self):
-        self.output_values = []
-        self.iteration = 0
+    def __init__(self, sfg: SFG, input_providers: Optional[Sequence[Optional[InputProvider]]] = None):
+        self._sfg = sfg
+        self._results = defaultdict(list)
+        self._delays = {}
+        self._iteration = 0
+        self._input_functions = [lambda _: 0 for _ in range(self._sfg.input_count)]
+        self._input_length = None
+        if input_providers is not None:
+            self.set_inputs(input_providers)
 
+    def set_input(self, index: int, input_provider: InputProvider) -> 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-{len(self._input_functions) - 1}, got {index})")
+        if callable(input_provider):
+            self._input_functions[index] = input_provider
+        elif isinstance(input_provider, Number):
+            self._input_functions[index] = lambda _: input_provider
+        else:
+            if self._input_length is None:
+                self._input_length = len(input_provider)
+            elif self._input_length != len(input_provider):
+                raise ValueError(f"Inconsistent input length for simulation (was {self._input_length}, got {len(input_provider)})")
+            self._input_functions[index] = lambda n: input_provider[n]
 
-class SimulationState:
-    """Simulation state.
-    TODO: More info.
-    """
+    def set_inputs(self, input_providers: Sequence[Optional[InputProvider]]) -> None:
+        """Set the input functions used to get values for the inputs to the internal SFG."""
+        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)})")
+        for index, input_provider in enumerate(input_providers):
+            if input_provider is not None:
+                self.set_input(index, input_provider)
+
+    def step(self, save_results: bool = True, bits_override: Optional[int] = None, truncate: bool = True) -> Sequence[Number]:
+        """Run one iteration of the simulation and return the resulting output values."""
+        return self.run_for(1, save_results, bits_override, truncate)
+
+    def run_until(self, iteration: int, save_results: bool = True, bits_override: Optional[int] = None, truncate: bool = True) -> Sequence[Number]:
+        """Run the simulation until its iteration is greater than or equal to the given iteration
+        and return the output values of the last iteration.
+        """
+        result = []
+        while self._iteration < iteration:
+            input_values = [self._input_functions[i](self._iteration) for i in range(self._sfg.input_count)]
+            results = {}
+            result = self._sfg.evaluate_outputs(input_values, results, self._delays, "", bits_override, truncate)
+            if save_results:
+                for key, value in results.items():
+                    self._results[key].append(value)
+            self._iteration += 1
+        return result
+
+    def run_for(self, iterations: int, save_results: bool = True, bits_override: Optional[int] = None, truncate: bool = True) -> Sequence[Number]:
+        """Run a given number of iterations of the simulation and return the output values of the last iteration."""
+        return self.run_until(self._iteration + iterations, save_results, bits_override, truncate)
+
+    def run(self, save_results: bool = True, bits_override: Optional[int] = None, truncate: bool = True) -> Sequence[Number]:
+        """Run the simulation until the end of its input arrays and return the output values of the last iteration."""
+        if self._input_length is None:
+            raise IndexError("Tried to run unlimited simulation")
+        return self.run_until(self._input_length, save_results, bits_override, truncate)
+
+    @property
+    def iteration(self) -> int:
+        """Get the current iteration number of the simulation."""
+        return self._iteration
 
-    # operation_states: Dict[OperationId, OperationState]
-    iteration: int
+    @property
+    def results(self) -> ResultArrayMap:
+        """Get a mapping from result keys to numpy arrays containing all results, including intermediate values,
+        calculated for each iteration up until now that was run with save_results enabled.
+        The mapping is indexed using the key() method of Operation with the appropriate output index.
+        Example result after 3 iterations: {"c1": [3, 6, 7], "c2": [4, 5, 5], "bfly1.0": [7, 0, 0], "bfly1.1": [-1, 0, 2], "0": [7, -2, -1]}
+        """
+        return {key: np.array(value) for key, value in self._results.items()}
 
-    def __init__(self):
-        self.operation_states = {}
-        self.iteration = 0
+    def clear_results(self) -> None:
+        """Clear all results that were saved until now."""
+        self._results.clear()
 
-    # TODO: More stuff.
+    def clear_state(self) -> None:
+        """Clear all current state of the simulation, except for the results and iteration."""
+        self._delays.clear()

b_asic/special_operations.py

+"""@package docstring
+B-ASIC Special Operations Module.
+TODO: More info.
+"""
+
+from numbers import Number
+from typing import Optional, Sequence
+
+from b_asic.operation import AbstractOperation, ResultKey, DelayMap, MutableResultMap, MutableDelayMap
+from b_asic.graph_component import Name, TypeName
+from b_asic.port import SignalSourceProvider
+
+
+class Input(AbstractOperation):
+    """Input operation.
+    TODO: More info.
+    """
+
+    def __init__(self, name: Name = ""):
+        super().__init__(input_count=0, output_count=1, name=name)
+        self.set_param("value", 0)
+
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "in"
+
+    def evaluate(self):
+        return self.param("value")
+
+    @property
+    def value(self) -> Number:
+        """Get the current value of this input."""
+        return self.param("value")
+
+    @value.setter
+    def value(self, value: Number) -> None:
+        """Set the current value of this input."""
+        self.set_param("value", value)
+
+
+class Output(AbstractOperation):
+    """Output operation.
+    TODO: More info.
+    """
+
+    def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
+        super().__init__(input_count=1, output_count=0,
+                         name=name, input_sources=[src0])
+
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "out"
+
+    def evaluate(self, _):
+        return None
+
+
+class Delay(AbstractOperation):
+    """Unit delay operation.
+    TODO: More info.
+    """
+
+    def __init__(self, src0: Optional[SignalSourceProvider] = None, initial_value: Number = 0, name: Name = ""):
+        super().__init__(input_count=1, output_count=1,
+                         name=name, input_sources=[src0])
+        self.set_param("initial_value", initial_value)
+
+    @classmethod
+    def type_name(cls) -> TypeName:
+        return "t"
+
+    def evaluate(self, a):
+        return self.param("initial_value")
+
+    def current_output(self, index: int, delays: Optional[DelayMap] = None, prefix: str = "") -> Optional[Number]:
+        if delays is not None:
+            return delays.get(self.key(index, prefix), self.param("initial_value"))
+        return self.param("initial_value")
+    
+    def evaluate_output(self, index: int, input_values: Sequence[Number], results: Optional[MutableResultMap] = None, delays: Optional[MutableDelayMap] = None, prefix: str = "", bits_override: Optional[int] = None, truncate: bool = True) -> Number:
+        if index != 0:
+            raise IndexError(
+                f"Output index out of range (expected 0-0, got {index})")
+        if len(input_values) != 1:
+            raise ValueError(
+                f"Wrong number of inputs supplied to SFG for evaluation (expected 1, got {len(input_values)})")
+
+        key = self.key(index, prefix)
+        value = self.param("initial_value")
+        if delays is not None:
+            value = delays.get(key, value)
+            delays[key] = self.truncate_inputs(input_values, bits_override)[0] if truncate else input_values[0]
+        if results is not None:
+            results[key] = value
+        return value
+
+    @property
+    def initial_value(self) -> Number:
+        """Get the initial value of this delay."""
+        return self.param("initial_value")
+
+    @initial_value.setter
+    def initial_value(self, value: Number) -> None:
+        """Set the initial value of this delay."""
+        self.set_param("initial_value", value)

legacy/README.md

+# Legacy files
+
+This folder contains currently unused code that is kept for acedemic purposes,
+or to be used as a refererence for future development.
+
+## simulation_oop
+
+This folder contains a C++ implementation of the Simulation class designed
+using Object-Oriented Programming, as opposed to the current version that uses
+Data-Oriented Design. They are functionally identical, but use different
+styles of programming and have different performance characteristics.
\ No newline at end of file