diff --git a/b_asic/graph_component.py b/b_asic/graph_component.py
index 0092e1b6ed8b5e3c3acd591910ce50c8aeb152c8..e37997016a3276f4dbde0394c44bf5c54ecbd51c 100644
--- a/b_asic/graph_component.py
+++ b/b_asic/graph_component.py
@@ -6,7 +6,7 @@ TODO: More info.
from abc import ABC, abstractmethod
from collections import deque
from copy import copy, deepcopy
-from typing import NewType, Any, Optional, Dict, Iterable, Generator
+from typing import NewType, Any, Dict, Mapping, Iterable, Generator
Name = NewType("Name", str)
@@ -52,13 +52,14 @@ class GraphComponent(ABC):
and that a new local id will be generated for it instead."""
raise NotImplementedError
+ @property
@abstractmethod
- def params(self) -> Dict[str, Optional[Any]]:
+ def params(self) -> Mapping[str, Any]:
"""Get a dictionary of all parameter values."""
raise NotImplementedError
@abstractmethod
- def param(self, name: str) -> Optional[Any]:
+ def param(self, name: str) -> Any:
"""Get the value of a parameter.
Returns None if the parameter is not defined.
"""
@@ -97,7 +98,7 @@ class AbstractGraphComponent(GraphComponent):
_name: Name
_graph_id: GraphID
- _parameters: Dict[str, Optional[Any]]
+ _parameters: Dict[str, Any]
def __init__(self, name: Name = ""):
self._name = name
@@ -121,10 +122,10 @@ class AbstractGraphComponent(GraphComponent):
self._graph_id = graph_id
@property
- def params(self) -> Dict[str, Optional[Any]]:
+ def params(self) -> Mapping[str, Any]:
return self._parameters.copy()
- def param(self, name: str) -> Optional[Any]:
+ def param(self, name: str) -> Any:
return self._parameters.get(name)
def set_param(self, name: str, value: Any) -> None:
diff --git a/b_asic/operation.py b/b_asic/operation.py
index ed0e7dfd1deb8bb217417eb71ad21cb0287a3bce..c75ca87b96d8794a4f3eb6ee939c446d3f02b2df 100644
--- a/b_asic/operation.py
+++ b/b_asic/operation.py
@@ -15,6 +15,10 @@ from b_asic.port import SignalSourceProvider, InputPort, OutputPort
from b_asic.signal import Signal
ResultKey = NewType("ResultKey", str)
+ResultMap = Mapping[ResultKey, Optional[Number]]
+MutableResultMap = MutableMapping[ResultKey, Optional[Number]]
+RegisterMap = Mapping[ResultKey, Number]
+MutableRegisterMap = MutableMapping[ResultKey, Number]
class Operation(GraphComponent, SignalSourceProvider):
"""Operation interface.
@@ -81,36 +85,36 @@ class Operation(GraphComponent, SignalSourceProvider):
@property
@abstractmethod
- def inputs(self) -> Sequence[InputPort]:
- """Get all input ports."""
+ def input_count(self) -> int:
+ """Get the number of input ports."""
raise NotImplementedError
@property
@abstractmethod
- def outputs(self) -> Sequence[OutputPort]:
- """Get all output ports."""
+ def output_count(self) -> int:
+ """Get the number of output ports."""
raise NotImplementedError
- @property
@abstractmethod
- def input_count(self) -> int:
- """Get the number of input ports."""
+ def input(self, index: int) -> InputPort:
+ """Get the input port at the given index."""
raise NotImplementedError
- @property
@abstractmethod
- def output_count(self) -> int:
- """Get the number of output ports."""
+ def output(self, index: int) -> OutputPort:
+ """Get the output port at the given index."""
raise NotImplementedError
+ @property
@abstractmethod
- def input(self, i: int) -> InputPort:
- """Get the input port at index i."""
+ def inputs(self) -> Sequence[InputPort]:
+ """Get all input ports."""
raise NotImplementedError
+ @property
@abstractmethod
- def output(self, i: int) -> OutputPort:
- """Get the output port at index i."""
+ def outputs(self) -> Sequence[OutputPort]:
+ """Get all output ports."""
raise NotImplementedError
@property
@@ -137,9 +141,8 @@ class Operation(GraphComponent, SignalSourceProvider):
raise NotImplementedError
@abstractmethod
- def current_output(self, index: int, results: Optional[MutableMapping[ResultKey, Optional[Number]]] = None, registers: Optional[Mapping[ResultKey, Number]] = None, prefix: str = "") -> Optional[Number]:
+ def current_output(self, index: int, registers: Optional[RegisterMap] = None, prefix: str = "") -> Optional[Number]:
"""Get the current output at the given index of this operation, if available.
- The results parameter will be used to store any results (including intermediate results) for caching.
The registers parameter will be used for lookup.
The prefix parameter will be used as a prefix for the key string when looking for registers.
See also: current_outputs, evaluate_output, evaluate_outputs.
@@ -147,7 +150,7 @@ class Operation(GraphComponent, SignalSourceProvider):
raise NotImplementedError
@abstractmethod
- def evaluate_output(self, index: int, input_values: Sequence[Number], results: Optional[MutableMapping[ResultKey, Optional[Number]]] = None, registers: Optional[MutableMapping[ResultKey, Number]] = None, prefix: str = "") -> Number:
+ def evaluate_output(self, index: int, input_values: Sequence[Number], results: Optional[MutableResultMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = "") -> Number:
"""Evaluate the output at the given index of this operation with the given input values.
The results parameter will be used to store any results (including intermediate results) for caching.
The registers parameter will be used to get the current value of any intermediate registers that are encountered, and be updated with their new values.
@@ -157,14 +160,14 @@ class Operation(GraphComponent, SignalSourceProvider):
raise NotImplementedError
@abstractmethod
- def current_outputs(self, results: Optional[MutableMapping[ResultKey, Optional[Number]]] = None, registers: Optional[Mapping[ResultKey, Number]] = None, prefix: str = "") -> Sequence[Optional[Number]]:
+ def current_outputs(self, registers: Optional[RegisterMap] = None, prefix: str = "") -> Sequence[Optional[Number]]:
"""Get all current outputs of this operation, if available.
See current_output for more information.
"""
raise NotImplementedError
@abstractmethod
- def evaluate_outputs(self, input_values: Sequence[Number], results: MutableMapping[ResultKey, Number], registers: MutableMapping[ResultKey, Number], prefix: str = "") -> Sequence[Number]:
+ def evaluate_outputs(self, input_values: Sequence[Number], results: Optional[MutableResultMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = "") -> Sequence[Number]:
"""Evaluate all outputs of this operation given the input values.
See evaluate_output for more information.
"""
@@ -200,29 +203,6 @@ class AbstractOperation(Operation, AbstractGraphComponent):
if src is not None:
self._input_ports[i].connect(src.source)
- def truncate_input(self, index: int, value: Number, bits: int) -> Number:
- """Truncate the value to be used as input at the given index to a certain bit length."""
- n = value
- if not isinstance(n, int):
- n = trunc(value)
- return n & ((2 ** bits) - 1)
-
- def truncate_inputs(self, input_values: Sequence[Number]) -> Sequence[Number]:
- """Truncate the values to be used as inputs to the bit lengths specified by the respective signals connected to each input."""
- args = []
- for i, input_port in enumerate(self.inputs):
- if input_port.signal_count >= 1:
- bits = input_port.signals[0].bits
- if bits is None:
- args.append(input_values[i])
- else:
- if isinstance(input_values[i], complex):
- raise TypeError("Complex value cannot be truncated to {bits} bits as requested by the signal connected to input #{i}")
- args.append(self.truncate_input(i, input_values[i], bits))
- else:
- args.append(input_values[i])
- return args
-
@abstractmethod
def evaluate(self, *inputs) -> Any: # pylint: disable=arguments-differ
"""Evaluate the operation and generate a list of output values given a list of input values."""
@@ -259,15 +239,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
def __rtruediv__(self, src: Union[SignalSourceProvider, Number]) -> "Division":
from b_asic.core_operations import Constant, Division # Import here to avoid circular imports.
return Division(Constant(src) if isinstance(src, Number) else src, self)
-
- @property
- def inputs(self) -> Sequence[InputPort]:
- return self._input_ports.copy()
-
- @property
- def outputs(self) -> Sequence[OutputPort]:
- return self._output_ports.copy()
-
+
@property
def input_count(self) -> int:
return len(self._input_ports)
@@ -276,11 +248,19 @@ class AbstractOperation(Operation, AbstractGraphComponent):
def output_count(self) -> int:
return len(self._output_ports)
- def input(self, i: int) -> InputPort:
- return self._input_ports[i]
+ def input(self, index: int) -> InputPort:
+ return self._input_ports[index]
+
+ def output(self, index: int) -> OutputPort:
+ return self._output_ports[index]
+
+ @property
+ def inputs(self) -> Sequence[InputPort]:
+ return self._input_ports
- def output(self, i: int) -> OutputPort:
- return self._output_ports[i]
+ @property
+ def outputs(self) -> Sequence[OutputPort]:
+ return self._output_ports
@property
def input_signals(self) -> Iterable[Signal]:
@@ -308,12 +288,10 @@ class AbstractOperation(Operation, AbstractGraphComponent):
key = str(index)
return key
- def current_output(self, index: int, results: Optional[MutableMapping[ResultKey, Optional[Number]]] = None, registers: Optional[Mapping[ResultKey, Number]] = None, prefix: str = "") -> Optional[Number]:
- if results is not None:
- results[self.key(index, prefix)] = None
+ def current_output(self, index: int, registers: Optional[RegisterMap] = None, prefix: str = "") -> Optional[Number]:
return None
- def evaluate_output(self, index: int, input_values: Sequence[Number], results: Optional[MutableMapping[ResultKey, Optional[Number]]] = None, registers: Optional[MutableMapping[ResultKey, Number]] = None, prefix: str = "") -> Number:
+ def evaluate_output(self, index: int, input_values: Sequence[Number], results: Optional[MutableResultMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = "") -> Number:
if index < 0 or index >= self.output_count:
raise IndexError(f"Output index out of range (expected 0-{self.output_count - 1}, got {index})")
if len(input_values) != self.input_count:
@@ -341,10 +319,10 @@ class AbstractOperation(Operation, AbstractGraphComponent):
results[self.key(i, prefix)] = values[i]
return values[index]
- def current_outputs(self, results: Optional[MutableMapping[ResultKey, Optional[Number]]] = None, registers: Optional[Mapping[ResultKey, Number]] = None, prefix: str = "") -> Sequence[Optional[Number]]:
- return [self.current_output(i, results, registers, prefix) for i in range(self.output_count)]
+ def current_outputs(self, registers: Optional[RegisterMap] = None, prefix: str = "") -> Sequence[Optional[Number]]:
+ return [self.current_output(i, registers, prefix) for i in range(self.output_count)]
- def evaluate_outputs(self, input_values: Sequence[Number], results: MutableMapping[ResultKey, Number], registers: MutableMapping[ResultKey, Number], prefix: str = "") -> Sequence[Number]:
+ def evaluate_outputs(self, input_values: Sequence[Number], results: Optional[MutableResultMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = "") -> Sequence[Number]:
return [self.evaluate_output(i, input_values, results, registers, prefix) for i in range(self.output_count)]
def split(self) -> Iterable[Operation]:
@@ -371,4 +349,27 @@ class AbstractOperation(Operation, AbstractGraphComponent):
if self.output_count != 1:
diff = "more" if self.output_count > 1 else "less"
raise TypeError(f"{self.__class__.__name__} cannot be used as an input source because it has {diff} than 1 output")
- return self.output(0)
\ No newline at end of file
+ return self.output(0)
+
+ def truncate_input(self, index: int, value: Number, bits: int) -> Number:
+ """Truncate the value to be used as input at the given index to a certain bit length."""
+ n = value
+ if not isinstance(n, int):
+ n = trunc(value)
+ return n & ((2 ** bits) - 1)
+
+ def truncate_inputs(self, input_values: Sequence[Number]) -> Sequence[Number]:
+ """Truncate the values to be used as inputs to the bit lengths specified by the respective signals connected to each input."""
+ args = []
+ for i, input_port in enumerate(self.inputs):
+ if input_port.signal_count >= 1:
+ bits = input_port.signals[0].bits
+ if bits is None:
+ args.append(input_values[i])
+ else:
+ if isinstance(input_values[i], complex):
+ raise TypeError("Complex value cannot be truncated to {bits} bits as requested by the signal connected to input #{i}")
+ args.append(self.truncate_input(i, input_values[i], bits))
+ else:
+ args.append(input_values[i])
+ return args
\ No newline at end of file
diff --git a/b_asic/signal_flow_graph.py b/b_asic/signal_flow_graph.py
index 4b5d2450b4ea350489862e6b06cd951940f12d54..573fb21c3e0c80a72de06af3621d87bcef7c0d5d 100644
--- a/b_asic/signal_flow_graph.py
+++ b/b_asic/signal_flow_graph.py
@@ -3,12 +3,12 @@ B-ASIC Signal Flow Graph Module.
TODO: More info.
"""
-from typing import List, Iterable, Sequence, Dict, MutableMapping, Optional, DefaultDict, Set
+from typing import List, Iterable, Sequence, Dict, Optional, DefaultDict, Set
from numbers import Number
from collections import defaultdict, deque
from b_asic.port import SignalSourceProvider, OutputPort
-from b_asic.operation import Operation, AbstractOperation, ResultKey
+from b_asic.operation import Operation, AbstractOperation, ResultKey, RegisterMap, MutableResultMap, MutableRegisterMap
from b_asic.signal import Signal
from b_asic.graph_component import GraphID, GraphIDNumber, GraphComponent, Name, TypeName
from b_asic.special_operations import Input, Output
@@ -49,15 +49,17 @@ class SFG(AbstractOperation):
_original_input_signals_to_indices: Dict[Signal, int]
_original_output_signals_to_indices: Dict[Signal, int]
- def __init__(self, input_signals: Sequence[Signal] = [], output_signals: Sequence[Signal] = [], \
- inputs: Sequence[Input] = [], outputs: Sequence[Output] = [], \
+ def __init__(self, input_signals: Optional[Sequence[Signal]] = None, output_signals: Optional[Sequence[Signal]] = None, \
+ inputs: Optional[Sequence[Input]] = None, outputs: Optional[Sequence[Output]] = None, \
id_number_offset: GraphIDNumber = 0, name: Name = "", \
input_sources: Optional[Sequence[Optional[SignalSourceProvider]]] = None):
- super().__init__(
- input_count=len(input_signals) + len(inputs),
- output_count=len(output_signals) + len(outputs),
- name=name,
- input_sources=input_sources)
+ input_signal_count = 0 if input_signals is None else len(input_signals)
+ input_operation_count = 0 if inputs is None else len(inputs)
+ output_signal_count = 0 if output_signals is None else len(output_signals)
+ output_operation_count = 0 if outputs is None else len(outputs)
+ super().__init__(input_count = input_signal_count + input_operation_count,
+ output_count = output_signal_count + output_operation_count,
+ name = name, input_sources = input_sources)
self._components_by_id = dict()
self._components_by_name = defaultdict(list)
@@ -71,58 +73,62 @@ class SFG(AbstractOperation):
self._original_output_signals_to_indices = {}
# Setup input signals.
- for input_index, signal in enumerate(input_signals):
- assert signal not in self._original_components_to_new, "Duplicate input signals supplied to SFG construcctor."
- new_input_op = self._add_component_copy_unconnected(Input())
- new_signal = self._add_component_copy_unconnected(signal)
- new_signal.set_source(new_input_op.output(0))
- self._input_operations.append(new_input_op)
- self._original_input_signals_to_indices[signal] = input_index
-
- # Setup input operations, starting from indices ater input signals.
- for input_index, input_op in enumerate(inputs, len(input_signals)):
- assert input_op not in self._original_components_to_new, "Duplicate input operations supplied to SFG constructor."
- new_input_op = self._add_component_copy_unconnected(input_op)
- for signal in input_op.output(0).signals:
- assert signal not in self._original_components_to_new, "Duplicate input signals connected to input ports supplied to SFG construcctor."
- new_signal = self._add_component_copy_unconnected(signal)
+ if input_signals is not None:
+ for input_index, signal in enumerate(input_signals):
+ assert signal not in self._original_components_to_new, "Duplicate input signals supplied to SFG construcctor."
+ new_input_op = self._add_component_unconnected_copy(Input())
+ new_signal = self._add_component_unconnected_copy(signal)
new_signal.set_source(new_input_op.output(0))
+ self._input_operations.append(new_input_op)
self._original_input_signals_to_indices[signal] = input_index
- self._input_operations.append(new_input_op)
+ # Setup input operations, starting from indices ater input signals.
+ if inputs is not None:
+ for input_index, input_op in enumerate(inputs, input_signal_count):
+ assert input_op not in self._original_components_to_new, "Duplicate input operations supplied to SFG constructor."
+ new_input_op = self._add_component_unconnected_copy(input_op)
+ for signal in input_op.output(0).signals:
+ assert signal not in self._original_components_to_new, "Duplicate input signals connected to input ports supplied to SFG construcctor."
+ new_signal = self._add_component_unconnected_copy(signal)
+ new_signal.set_source(new_input_op.output(0))
+ self._original_input_signals_to_indices[signal] = input_index
+
+ self._input_operations.append(new_input_op)
# Setup output signals.
- for output_index, signal in enumerate(output_signals):
- new_output_op = self._add_component_copy_unconnected(Output())
- if signal in self._original_components_to_new:
- # Signal was already added when setting up inputs.
- new_signal = self._original_components_to_new[signal]
- new_signal.set_destination(new_output_op.input(0))
- else:
- # New signal has to be created.
- new_signal = self._add_component_copy_unconnected(signal)
- new_signal.set_destination(new_output_op.input(0))
-
- self._output_operations.append(new_output_op)
- self._original_output_signals_to_indices[signal] = output_index
-
- # Setup output operations, starting from indices after output signals.
- for output_index, output_op in enumerate(outputs, len(output_signals)):
- assert output_op not in self._original_components_to_new, "Duplicate output operations supplied to SFG constructor."
- new_output_op = self._add_component_copy_unconnected(output_op)
- for signal in output_op.input(0).signals:
- new_signal = None
+ if output_signals is not None:
+ for output_index, signal in enumerate(output_signals):
+ new_output_op = self._add_component_unconnected_copy(Output())
if signal in self._original_components_to_new:
# Signal was already added when setting up inputs.
new_signal = self._original_components_to_new[signal]
+ new_signal.set_destination(new_output_op.input(0))
else:
# New signal has to be created.
- new_signal = self._add_component_copy_unconnected(signal)
+ new_signal = self._add_component_unconnected_copy(signal)
+ new_signal.set_destination(new_output_op.input(0))
- new_signal.set_destination(new_output_op.input(0))
+ self._output_operations.append(new_output_op)
self._original_output_signals_to_indices[signal] = output_index
- self._output_operations.append(new_output_op)
+ # Setup output operations, starting from indices after output signals.
+ if outputs is not None:
+ for output_index, output_op in enumerate(outputs, output_signal_count):
+ assert output_op not in self._original_components_to_new, "Duplicate output operations supplied to SFG constructor."
+ new_output_op = self._add_component_unconnected_copy(output_op)
+ for signal in output_op.input(0).signals:
+ new_signal = None
+ if signal in self._original_components_to_new:
+ # Signal was already added when setting up inputs.
+ new_signal = self._original_components_to_new[signal]
+ else:
+ # New signal has to be created.
+ new_signal = self._add_component_unconnected_copy(signal)
+
+ new_signal.set_destination(new_output_op.input(0))
+ self._original_output_signals_to_indices[signal] = output_index
+
+ self._output_operations.append(new_output_op)
output_operations_set = set(self._output_operations)
@@ -133,13 +139,12 @@ class SFG(AbstractOperation):
if new_signal.destination is None:
if signal.destination is None:
raise ValueError(f"Input signal #{input_index} is missing destination in SFG")
- elif signal.destination.operation not in self._original_components_to_new:
- self._add_operation_copy_connected_tree(signal.destination.operation)
- else:
- if new_signal.destination.operation in output_operations_set:
- # Add directly connected input to output to ordered list.
- self._components_ordered.extend([new_signal.source.operation, new_signal, new_signal.destination.operation])
- self._operations_ordered.extend([new_signal.source.operation, new_signal.destination.operation])
+ if signal.destination.operation not in self._original_components_to_new:
+ self._add_operation_connected_tree_copy(signal.destination.operation)
+ elif new_signal.destination.operation in output_operations_set:
+ # Add directly connected input to output to ordered list.
+ self._components_ordered.extend([new_signal.source.operation, new_signal, new_signal.destination.operation])
+ self._operations_ordered.extend([new_signal.source.operation, new_signal.destination.operation])
# Search the graph inwards from each output signal.
for signal, output_index in self._original_output_signals_to_indices.items():
@@ -149,13 +154,12 @@ class SFG(AbstractOperation):
if signal.source is None:
raise ValueError(f"Output signal #{output_index} is missing source in SFG")
if signal.source.operation not in self._original_components_to_new:
- self._add_operation_copy_connected_tree(signal.source.operation)
+ self._add_operation_connected_tree_copy(signal.source.operation)
def __call__(self, *src: Optional[SignalSourceProvider], name: Name = "") -> "SFG":
"""Get a new independent SFG instance that is identical to this SFG except without any of its external connections."""
return SFG(inputs = self._input_operations, outputs = self._output_operations,
- id_number_offset = self._graph_id_generator.id_number_offset,
- name = name, input_sources = src if src else None)
+ id_number_offset = self.id_number_offset, name = name, input_sources = src if src else None)
@property
def type_name(self) -> TypeName:
@@ -166,7 +170,7 @@ class SFG(AbstractOperation):
n = len(result)
return None if n == 0 else result[0] if n == 1 else result
- def evaluate_output(self, index: int, input_values: Sequence[Number], results: Optional[MutableMapping[ResultKey, Optional[Number]]] = None, registers: Optional[MutableMapping[ResultKey, Number]] = None, prefix: str = "") -> Number:
+ def evaluate_output(self, index: int, input_values: Sequence[Number], results: Optional[MutableResultMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = "") -> Number:
if index < 0 or index >= self.output_count:
raise IndexError(f"Output index out of range (expected 0-{self.output_count - 1}, got {index})")
if len(input_values) != self.input_count:
@@ -189,7 +193,7 @@ class SFG(AbstractOperation):
def copy_component(self, *args, **kwargs) -> GraphComponent:
return super().copy_component(*args, **kwargs, inputs = self._input_operations, outputs = self._output_operations,
- id_number_offset = self._graph_id_generator.id_number_offset, name = self.name)
+ id_number_offset = self.id_number_offset, name = self.name)
@property
def id_number_offset(self) -> GraphIDNumber:
@@ -224,7 +228,7 @@ class SFG(AbstractOperation):
"""
return self._components_by_name.get(name, [])
- def _add_component_copy_unconnected(self, original_component: GraphComponent) -> GraphComponent:
+ def _add_component_unconnected_copy(self, original_component: GraphComponent) -> GraphComponent:
assert original_component not in self._original_components_to_new, "Tried to add duplicate SFG component"
new_component = original_component.copy_component()
self._original_components_to_new[original_component] = new_component
@@ -234,14 +238,14 @@ class SFG(AbstractOperation):
self._components_by_name[new_component.name].append(new_component)
return new_component
- def _add_operation_copy_connected_tree(self, start_op: Operation):
+ def _add_operation_connected_tree_copy(self, start_op: Operation) -> None:
op_stack = deque([start_op])
while op_stack:
original_op = op_stack.pop()
# Add or get the new copy of the operation.
new_op = None
if original_op not in self._original_components_to_new:
- new_op = self._add_component_copy_unconnected(original_op)
+ new_op = self._add_component_unconnected_copy(original_op)
self._components_ordered.append(new_op)
self._operations_ordered.append(new_op)
else:
@@ -266,7 +270,7 @@ class SFG(AbstractOperation):
if original_signal.source is None:
raise ValueError("Dangling signal without source in SFG")
- new_signal = self._add_component_copy_unconnected(original_signal)
+ new_signal = self._add_component_unconnected_copy(original_signal)
new_signal.set_destination(new_op.input(original_input_port.index))
self._components_ordered.append(new_signal)
@@ -277,7 +281,7 @@ class SFG(AbstractOperation):
new_signal.set_source(self._original_components_to_new[original_connected_op].output(original_signal.source.index))
else:
# Create new operation, set signal source to it.
- new_connected_op = self._add_component_copy_unconnected(original_connected_op)
+ new_connected_op = self._add_component_unconnected_copy(original_connected_op)
new_signal.set_source(new_connected_op.output(original_signal.source.index))
self._components_ordered.append(new_connected_op)
self._operations_ordered.append(new_connected_op)
@@ -301,7 +305,7 @@ class SFG(AbstractOperation):
if original_signal.source is None:
raise ValueError("Dangling signal without source in SFG")
- new_signal = self._add_component_copy_unconnected(original_signal)
+ new_signal = self._add_component_unconnected_copy(original_signal)
new_signal.set_source(new_op.output(original_output_port.index))
self._components_ordered.append(new_signal)
@@ -312,7 +316,7 @@ class SFG(AbstractOperation):
new_signal.set_destination(self._original_components_to_new[original_connected_op].input(original_signal.destination.index))
else:
# Create new operation, set destination to it.
- new_connected_op = self._add_component_copy_unconnected(original_connected_op)
+ new_connected_op = self._add_component_unconnected_copy(original_connected_op)
new_signal.set_destination(new_connected_op.input(original_signal.destination.index))
self._components_ordered.append(new_connected_op)
self._operations_ordered.append(new_connected_op)
@@ -320,7 +324,7 @@ class SFG(AbstractOperation):
# Add connected operation to the queue of operations to visit.
op_stack.append(original_connected_op)
- def _evaluate_source(self, src: OutputPort, results: MutableMapping[ResultKey, Number], registers: MutableMapping[ResultKey, Number], prefix: str) -> Number:
+ def _evaluate_source(self, src: OutputPort, results: MutableResultMap, registers: MutableRegisterMap, prefix: str) -> Number:
src_prefix = prefix
if src_prefix:
src_prefix += "."
@@ -333,7 +337,7 @@ class SFG(AbstractOperation):
raise RuntimeError(f"Direct feedback loop detected when evaluating operation.")
return value
- src.operation.current_output(src.index, results, registers, src_prefix)
+ results[key] = src.operation.current_output(src.index, registers, src_prefix)
input_values = [self._evaluate_source(input_port.signals[0].source, results, registers, prefix) for input_port in src.operation.inputs]
value = src.operation.evaluate_output(src.index, input_values, results, registers, src_prefix)
results[key] = value
diff --git a/b_asic/simulation.py b/b_asic/simulation.py
index b6d08bc1297deac1c5e1b9b8ec7945e72d66f811..9d0d154fa899923ee28cf444512262fb85c73a3a 100644
--- a/b_asic/simulation.py
+++ b/b_asic/simulation.py
@@ -7,10 +7,13 @@ from collections import defaultdict
from numbers import Number
from typing import List, Dict, DefaultDict, Callable, Sequence, Mapping, Union, Optional
-from b_asic.operation import ResultKey
+from b_asic.operation import ResultKey, ResultMap
from b_asic.signal_flow_graph import SFG
+InputProvider = Union[Number, Sequence[Number], Callable[[int], Number]]
+
+
class Simulation:
"""Simulation.
TODO: More info.
@@ -25,36 +28,37 @@ class Simulation:
_latest_output_values: Sequence[Number]
_save_results: bool
- def __init__(self, sfg: SFG, input_providers: Optional[Sequence[Union[None, Sequence[Number], Callable[[int], Number]]]] = None, save_results: bool = False):
+ def __init__(self, sfg: SFG, input_providers: Optional[Sequence[Optional[InputProvider]]] = None, save_results: bool = False):
self._sfg = sfg
self._results = defaultdict(dict)
self._registers = {}
self._iteration = 0
- self._input_functions = [lambda n: 0 for _ in range(self._sfg.input_count)]
- self._current_input_values = []
+ self._input_functions = [lambda _: 0 for _ in range(self._sfg.input_count)]
+ self._current_input_values = [0 for _ in range(self._sfg.input_count)]
self._latest_output_values = [0 for _ in range(self._sfg.output_count)]
self._save_results = save_results
if input_providers is not None:
self.set_inputs(input_providers)
- def set_inputs(self, input_providers: Sequence[Union[None, Sequence[Number], Callable[[int], Number]]]) -> None:
+ 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:
+ self._input_functions[index] = lambda n: input_provider[n]
+
+ 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)})")
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 set_input(self, index: int, input_provider: Union[None, 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-{len(self._input_functions) - 1}, got {index})")
-
- if input_provider is not None:
- if callable(input_provider):
- self._input_functions[index] = input_provider
- else:
- self._input_functions[index] = lambda n: input_provider[n]
+ if input_provider is not None:
+ self.set_input(index, input_provider)
@property
def save_results(self) -> bool:
@@ -91,7 +95,7 @@ class Simulation:
return self._iteration
@property
- def results(self) -> Mapping[int, Mapping[ResultKey, Number]]:
+ def results(self) -> Mapping[int, ResultMap]:
"""Get a mapping of all results, including intermediate values, calculated for each iteration up until now.
The outer mapping maps from iteration number to value mapping. The value mapping maps output port identifiers to values.
Example: {0: {"c1": 3, "c2": 4, "bfly1.0": 7, "bfly1.1": -1, "0": 7}}
@@ -100,4 +104,10 @@ class Simulation:
def clear_results(self) -> None:
"""Clear all results that were saved until now."""
- self._results.clear()
\ No newline at end of file
+ self._results.clear()
+
+ def clear_state(self) -> None:
+ """Clear all current state of the simulation, except for the results and iteration."""
+ self._registers.clear()
+ self._current_input_values = [0 for _ in range(self._sfg.input_count)]
+ self._latest_output_values = [0 for _ in range(self._sfg.output_count)]
\ No newline at end of file
diff --git a/b_asic/special_operations.py b/b_asic/special_operations.py
index 83555482daf8671fb18ee6a3a3b5139c18bc0ac3..ca7ff203d5d58b18b50dcfece6eeb0c036d993e9 100644
--- a/b_asic/special_operations.py
+++ b/b_asic/special_operations.py
@@ -4,9 +4,9 @@ TODO: More info.
"""
from numbers import Number
-from typing import Optional, Sequence, Mapping, MutableMapping
+from typing import Optional, Sequence
-from b_asic.operation import AbstractOperation, ResultKey
+from b_asic.operation import AbstractOperation, ResultKey, RegisterMap, MutableResultMap, MutableRegisterMap
from b_asic.graph_component import Name, TypeName
from b_asic.port import SignalSourceProvider
@@ -70,16 +70,12 @@ class Register(AbstractOperation):
def evaluate(self, a):
return self.param("initial_value")
- def current_output(self, index: int, results: Optional[MutableMapping[ResultKey, Optional[Number]]] = None, registers: Optional[Mapping[ResultKey, Number]] = None, prefix: str = "") -> Optional[Number]:
- key = self.key(index, prefix)
- value = self.param("initial_value")
+ def current_output(self, index: int, registers: Optional[RegisterMap] = None, prefix: str = "") -> Optional[Number]:
if registers is not None:
- value = registers.get(key, value)
- if results is not None:
- results[key] = value
- return value
+ return registers.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[MutableMapping[ResultKey, Optional[Number]]] = None, registers: Optional[MutableMapping[ResultKey, Number]] = None, prefix: str = ""):
+ def evaluate_output(self, index: int, input_values: Sequence[Number], results: Optional[MutableResultMap] = None, registers: Optional[MutableRegisterMap] = None, prefix: str = ""):
if index != 0:
raise IndexError(f"Output index out of range (expected 0-0, got {index})")
if len(input_values) != 1: