Skip to content
Snippets Groups Projects
Commit 7b732af1 authored by Jacob Wahlman's avatar Jacob Wahlman :ok_hand:
Browse files

Merge branch '26-print-pg' into 'develop' 

Resolve "Print PG"

See merge request PUM_TDDD96/B-ASIC!47
parents 6653c16f 9c9e2af6
Branches
Tags
3 merge requests!67WIP: B-ASIC version 1.0.0 hotfix,!65B-ASIC version 1.0.0,!47Resolve "Print PG"
Pipeline #15079 passed
Stage: test
Hide whitespace changes
Inline Side-by-side
b_asic/graph_component.py
+ 4
− 0
View file @ 7b732af1
...@@ -105,6 +105,10 @@ class AbstractGraphComponent(GraphComponent): ...@@ -105,6 +105,10 @@ class AbstractGraphComponent(GraphComponent):
self._graph_id = "" self._graph_id = ""
self._parameters = {} self._parameters = {}
def __str__(self):
return f"id: {self.graph_id if self.graph_id else 'no_id'}, \tname: {self.name if self.name else 'no_name'}" + \
"".join((f", \t{key}: {str(param)}" for key, param in self._parameters.items()))
@property @property
def name(self) -> Name: def name(self) -> Name:
return self._name return self._name
......
b_asic/operation.py
+ 51
− 0
View file @ 7b732af1
...@@ -262,6 +262,47 @@ class AbstractOperation(Operation, AbstractGraphComponent): ...@@ -262,6 +262,47 @@ class AbstractOperation(Operation, AbstractGraphComponent):
from b_asic.core_operations import Constant, Division from b_asic.core_operations import Constant, Division
return Division(Constant(src) if isinstance(src, Number) else src, self) return Division(Constant(src) if isinstance(src, Number) else src, self)
def __str__(self):
inputs_dict = dict()
for i, port in enumerate(self.inputs):
if port.signal_count == 0:
inputs_dict[i] = '-'
break
dict_ele = []
for signal in port.signals:
if signal.source:
if signal.source.operation.graph_id:
dict_ele.append(signal.source.operation.graph_id)
else:
dict_ele.append("no_id")
else:
if signal.graph_id:
dict_ele.append(signal.graph_id)
else:
dict_ele.append("no_id")
inputs_dict[i] = dict_ele
outputs_dict = dict()
for i, port in enumerate(self.outputs):
if port.signal_count == 0:
outputs_dict[i] = '-'
break
dict_ele = []
for signal in port.signals:
if signal.destination:
if signal.destination.operation.graph_id:
dict_ele.append(signal.destination.operation.graph_id)
else:
dict_ele.append("no_id")
else:
if signal.graph_id:
dict_ele.append(signal.graph_id)
else:
dict_ele.append("no_id")
outputs_dict[i] = dict_ele
return super().__str__() + f", \tinputs: {str(inputs_dict)}, \toutputs: {str(outputs_dict)}"
@property @property
def input_count(self) -> int: def input_count(self) -> int:
return len(self._input_ports) return len(self._input_ports)
...@@ -400,6 +441,16 @@ class AbstractOperation(Operation, AbstractGraphComponent): ...@@ -400,6 +441,16 @@ class AbstractOperation(Operation, AbstractGraphComponent):
def neighbors(self) -> Iterable[GraphComponent]: def neighbors(self) -> Iterable[GraphComponent]:
return list(self.input_signals) + list(self.output_signals) return list(self.input_signals) + list(self.output_signals)
@property
def preceding_operations(self) -> Iterable[Operation]:
"""Returns an Iterable of all Operations that are connected to this Operations input ports."""
return [signal.source.operation for signal in self.input_signals if signal.source]
@property
def subsequent_operations(self) -> Iterable[Operation]:
"""Returns an Iterable of all Operations that are connected to this Operations output ports."""
return [signal.destination.operation for signal in self.output_signals if signal.destination]
@property @property
def source(self) -> OutputPort: def source(self) -> OutputPort:
if self.output_count != 1: if self.output_count != 1:
......
b_asic/signal_flow_graph.py
+ 169
− 113
View file @ 7b732af1
...@@ -6,13 +6,15 @@ TODO: More info. ...@@ -6,13 +6,15 @@ TODO: More info.
from typing import List, Iterable, Sequence, Dict, Optional, DefaultDict, MutableSet from typing import List, Iterable, Sequence, Dict, Optional, DefaultDict, MutableSet
from numbers import Number from numbers import Number
from collections import defaultdict, deque from collections import defaultdict, deque
from io import StringIO
from queue import PriorityQueue
import itertools
from b_asic.port import SignalSourceProvider, OutputPort, InputPort from b_asic.port import SignalSourceProvider, OutputPort
from b_asic.operation import Operation, AbstractOperation, MutableOutputMap, MutableRegisterMap from b_asic.operation import Operation, AbstractOperation, MutableOutputMap, MutableRegisterMap
from b_asic.signal import Signal from b_asic.signal import Signal
from b_asic.graph_component import GraphID, GraphIDNumber, GraphComponent, Name, TypeName from b_asic.graph_component import GraphID, GraphIDNumber, GraphComponent, Name, TypeName
from b_asic.special_operations import Input, Output, Register from b_asic.special_operations import Input, Output, Register
from b_asic.core_operations import Constant
class GraphIDGenerator: class GraphIDGenerator:
...@@ -41,8 +43,9 @@ class SFG(AbstractOperation): ...@@ -41,8 +43,9 @@ class SFG(AbstractOperation):
_components_by_id: Dict[GraphID, GraphComponent] _components_by_id: Dict[GraphID, GraphComponent]
_components_by_name: DefaultDict[Name, List[GraphComponent]] _components_by_name: DefaultDict[Name, List[GraphComponent]]
_components_ordered: List[GraphComponent] _components_dfs_order: List[GraphComponent]
_operations_ordered: List[Operation] _operations_dfs_order: List[Operation]
_operations_topological_order: List[Operation]
_graph_id_generator: GraphIDGenerator _graph_id_generator: GraphIDGenerator
_input_operations: List[Input] _input_operations: List[Input]
_output_operations: List[Output] _output_operations: List[Output]
...@@ -67,8 +70,9 @@ class SFG(AbstractOperation): ...@@ -67,8 +70,9 @@ class SFG(AbstractOperation):
self._components_by_id = dict() self._components_by_id = dict()
self._components_by_name = defaultdict(list) self._components_by_name = defaultdict(list)
self._components_ordered = [] self._components_dfs_order = []
self._operations_ordered = [] self._operations_dfs_order = []
self._operations_topological_order = []
self._graph_id_generator = GraphIDGenerator(id_number_offset) self._graph_id_generator = GraphIDGenerator(id_number_offset)
self._input_operations = [] self._input_operations = []
self._output_operations = [] self._output_operations = []
...@@ -151,9 +155,9 @@ class SFG(AbstractOperation): ...@@ -151,9 +155,9 @@ class SFG(AbstractOperation):
signal.destination.operation) signal.destination.operation)
elif new_signal.destination.operation in output_operations_set: elif new_signal.destination.operation in output_operations_set:
# Add directly connected input to output to ordered list. # Add directly connected input to output to ordered list.
self._components_ordered.extend( self._components_dfs_order.extend(
[new_signal.source.operation, new_signal, new_signal.destination.operation]) [new_signal.source.operation, new_signal, new_signal.destination.operation])
self._operations_ordered.extend( self._operations_dfs_order.extend(
[new_signal.source.operation, new_signal.destination.operation]) [new_signal.source.operation, new_signal.destination.operation])
# Search the graph inwards from each output signal. # Search the graph inwards from each output signal.
...@@ -170,47 +174,18 @@ class SFG(AbstractOperation): ...@@ -170,47 +174,18 @@ class SFG(AbstractOperation):
def __str__(self) -> str: def __str__(self) -> str:
"""Get a string representation of this SFG.""" """Get a string representation of this SFG."""
output_string = "" string_io = StringIO()
for component in self._components_ordered: string_io.write(super().__str__() + "\n")
if isinstance(component, Operation): string_io.write("Internal Operations:\n")
for key, value in self._components_by_id.items(): line = "-" * 100 + "\n"
if value is component: string_io.write(line)
output_string += "id: " + key + ", name: "
if component.name != None:
output_string += component.name + ", "
else:
output_string += "-, "
if isinstance(component, Constant): for operation in self.get_operations_topological_order():
output_string += "value: " + \ string_io.write(str(operation) + "\n")
str(component.value) + ", input: ["
else: string_io.write(line)
output_string += "input: ["
return string_io.getvalue()
counter_input = 0
for input in component.inputs:
counter_input += 1
for signal in input.signals:
for key, value in self._components_by_id.items():
if value is signal:
output_string += key + ", "
if counter_input > 0:
output_string = output_string[:-2]
output_string += "], output: ["
counter_output = 0
for output in component.outputs:
counter_output += 1
for signal in output.signals:
for key, value in self._components_by_id.items():
if value is signal:
output_string += key + ", "
if counter_output > 0:
output_string = output_string[:-2]
output_string += "]\n"
return output_string
def __call__(self, *src: Optional[SignalSourceProvider], name: Name = "") -> "SFG": 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.""" """Get a new independent SFG instance that is identical to this SFG except without any of its external connections."""
...@@ -248,7 +223,7 @@ class SFG(AbstractOperation): ...@@ -248,7 +223,7 @@ class SFG(AbstractOperation):
return value return value
def connect_external_signals_to_components(self) -> bool: def connect_external_signals_to_components(self) -> bool:
""" Connects any external signals to this SFG's internal operations. This SFG becomes unconnected to the SFG """ Connects any external signals to this SFG's internal operations. This SFG becomes unconnected to the SFG
it is a component off, causing it to become invalid afterwards. Returns True if succesful, False otherwise. """ it is a component off, causing it to become invalid afterwards. Returns True if succesful, False otherwise. """
if len(self.inputs) != len(self.input_operations): if len(self.inputs) != len(self.input_operations):
raise IndexError(f"Number of inputs does not match the number of input_operations in SFG.") raise IndexError(f"Number of inputs does not match the number of input_operations in SFG.")
...@@ -264,7 +239,7 @@ class SFG(AbstractOperation): ...@@ -264,7 +239,7 @@ class SFG(AbstractOperation):
dest = input_operation.output(0).signals[0].destination dest = input_operation.output(0).signals[0].destination
dest.clear() dest.clear()
port.signals[0].set_destination(dest) port.signals[0].set_destination(dest)
# For each output_signal, connect it to the corresponding operation # For each output_signal, connect it to the corresponding operation
for port, output_operation in zip(self.outputs, self.output_operations): for port, output_operation in zip(self.outputs, self.output_operations):
src = output_operation.input(0).signals[0].source src = output_operation.input(0).signals[0].source
src.clear() src.clear()
...@@ -328,12 +303,12 @@ class SFG(AbstractOperation): ...@@ -328,12 +303,12 @@ class SFG(AbstractOperation):
@property @property
def components(self) -> Iterable[GraphComponent]: def components(self) -> Iterable[GraphComponent]:
"""Get all components of this graph in depth-first order.""" """Get all components of this graph in depth-first order."""
return self._components_ordered return self._components_dfs_order
@property @property
def operations(self) -> Iterable[Operation]: def operations(self) -> Iterable[Operation]:
"""Get all operations of this graph in depth-first order.""" """Get all operations of this graph in depth-first order."""
return self._operations_ordered return self._operations_dfs_order
def get_components_with_type_name(self, type_name: TypeName) -> List[GraphComponent]: def get_components_with_type_name(self, type_name: TypeName) -> List[GraphComponent]:
"""Get a list with all components in this graph with the specified type_name. """Get a list with all components in this graph with the specified type_name.
...@@ -387,8 +362,8 @@ class SFG(AbstractOperation): ...@@ -387,8 +362,8 @@ class SFG(AbstractOperation):
new_op = None new_op = None
if original_op not in self._original_components_to_new: if original_op not in self._original_components_to_new:
new_op = self._add_component_unconnected_copy(original_op) new_op = self._add_component_unconnected_copy(original_op)
self._components_ordered.append(new_op) self._components_dfs_order.append(new_op)
self._operations_ordered.append(new_op) self._operations_dfs_order.append(new_op)
else: else:
new_op = self._original_components_to_new[original_op] new_op = self._original_components_to_new[original_op]
...@@ -402,24 +377,20 @@ class SFG(AbstractOperation): ...@@ -402,24 +377,20 @@ class SFG(AbstractOperation):
if original_signal in self._original_input_signals_to_indices: if original_signal in self._original_input_signals_to_indices:
# New signal already created during first step of constructor. # New signal already created during first step of constructor.
new_signal = self._original_components_to_new[original_signal] new_signal = self._original_components_to_new[original_signal]
new_signal.set_destination( new_signal.set_destination(new_op.input(original_input_port.index))
new_op.input(original_input_port.index))
self._components_ordered.extend( self._components_dfs_order.extend([new_signal, new_signal.source.operation])
[new_signal, new_signal.source.operation]) self._operations_dfs_order.append(new_signal.source.operation)
self._operations_ordered.append(
new_signal.source.operation)
# Check if the signal has not been added before. # Check if the signal has not been added before.
elif original_signal not in self._original_components_to_new: elif original_signal not in self._original_components_to_new:
if original_signal.source is None: if original_signal.source is None:
raise ValueError( raise ValueError("Dangling signal without source in SFG")
"Dangling signal without source in SFG")
new_signal = self._add_component_unconnected_copy( new_signal = self._add_component_unconnected_copy(original_signal)
original_signal) new_signal.set_destination(new_op.input(original_input_port.index))
new_signal.set_destination(
new_op.input(original_input_port.index)) self._components_dfs_order.append(new_signal)
self._components_ordered.append(new_signal)
original_connected_op = original_signal.source.operation original_connected_op = original_signal.source.operation
# Check if connected Operation has been added before. # Check if connected Operation has been added before.
...@@ -429,12 +400,11 @@ class SFG(AbstractOperation): ...@@ -429,12 +400,11 @@ class SFG(AbstractOperation):
original_signal.source.index)) original_signal.source.index))
else: else:
# Create new operation, set signal source to it. # Create new operation, set signal source to it.
new_connected_op = self._add_component_unconnected_copy( new_connected_op = self._add_component_unconnected_copy(original_connected_op)
original_connected_op) new_signal.set_source(new_connected_op.output(original_signal.source.index))
new_signal.set_source(new_connected_op.output(
original_signal.source.index)) self._components_dfs_order.append(new_connected_op)
self._components_ordered.append(new_connected_op) self._operations_dfs_order.append(new_connected_op)
self._operations_ordered.append(new_connected_op)
# Add connected operation to queue of operations to visit. # Add connected operation to queue of operations to visit.
op_stack.append(original_connected_op) op_stack.append(original_connected_op)
...@@ -446,24 +416,20 @@ class SFG(AbstractOperation): ...@@ -446,24 +416,20 @@ class SFG(AbstractOperation):
if original_signal in self._original_output_signals_to_indices: if original_signal in self._original_output_signals_to_indices:
# New signal already created during first step of constructor. # New signal already created during first step of constructor.
new_signal = self._original_components_to_new[original_signal] new_signal = self._original_components_to_new[original_signal]
new_signal.set_source( new_signal.set_source(new_op.output(original_output_port.index))
new_op.output(original_output_port.index))
self._components_ordered.extend( self._components_dfs_order.extend([new_signal, new_signal.destination.operation])
[new_signal, new_signal.destination.operation]) self._operations_dfs_order.append(new_signal.destination.operation)
self._operations_ordered.append(
new_signal.destination.operation)
# Check if signal has not been added before. # Check if signal has not been added before.
elif original_signal not in self._original_components_to_new: elif original_signal not in self._original_components_to_new:
if original_signal.source is None: if original_signal.source is None:
raise ValueError( raise ValueError("Dangling signal without source in SFG")
"Dangling signal without source in SFG")
new_signal = self._add_component_unconnected_copy( new_signal = self._add_component_unconnected_copy(original_signal)
original_signal) new_signal.set_source(new_op.output(original_output_port.index))
new_signal.set_source(
new_op.output(original_output_port.index)) self._components_dfs_order.append(new_signal)
self._components_ordered.append(new_signal)
original_connected_op = original_signal.destination.operation original_connected_op = original_signal.destination.operation
# Check if connected operation has been added. # Check if connected operation has been added.
...@@ -473,12 +439,11 @@ class SFG(AbstractOperation): ...@@ -473,12 +439,11 @@ class SFG(AbstractOperation):
original_signal.destination.index)) original_signal.destination.index))
else: else:
# Create new operation, set destination to it. # Create new operation, set destination to it.
new_connected_op = self._add_component_unconnected_copy( new_connected_op = self._add_component_unconnected_copy(original_connected_op)
original_connected_op) new_signal.set_destination(new_connected_op.input(original_signal.destination.index))
new_signal.set_destination(new_connected_op.input(
original_signal.destination.index)) self._components_dfs_order.append(new_connected_op)
self._components_ordered.append(new_connected_op) self._operations_dfs_order.append(new_connected_op)
self._operations_ordered.append(new_connected_op)
# Add connected operation to the queue of operations to visit. # Add connected operation to the queue of operations to visit.
op_stack.append(original_connected_op) op_stack.append(original_connected_op)
...@@ -556,16 +521,13 @@ class SFG(AbstractOperation): ...@@ -556,16 +521,13 @@ class SFG(AbstractOperation):
if key in results: if key in results:
value = results[key] value = results[key]
if value is None: if value is None:
raise RuntimeError( raise RuntimeError(f"Direct feedback loop detected when evaluating operation.")
f"Direct feedback loop detected when evaluating operation.")
return value return value
results[key] = src.operation.current_output( results[key] = src.operation.current_output(src.index, registers, src_prefix)
src.index, registers, src_prefix)
input_values = [self._evaluate_source( input_values = [self._evaluate_source(
input_port.signals[0].source, results, registers, prefix) for input_port in src.operation.inputs] input_port.signals[0].source, results, registers, prefix) for input_port in src.operation.inputs]
value = src.operation.evaluate_output( value = src.operation.evaluate_output(src.index, input_values, results, registers, src_prefix)
src.index, input_values, results, registers, src_prefix)
results[key] = value results[key] = value
return value return value
...@@ -573,7 +535,7 @@ class SFG(AbstractOperation): ...@@ -573,7 +535,7 @@ class SFG(AbstractOperation):
"""Returns a Precedence list of the SFG where each element in n:th the list consists """Returns a Precedence list of the SFG where each element in n:th the list consists
of elements that are executed in the n:th step. If the precedence list already has been of elements that are executed in the n:th step. If the precedence list already has been
calculated for the current SFG then returns the cached version.""" calculated for the current SFG then returns the cached version."""
if self._precedence_list is not None: if self._precedence_list:
return self._precedence_list return self._precedence_list
# Find all operations with only outputs and no inputs. # Find all operations with only outputs and no inputs.
...@@ -587,17 +549,9 @@ class SFG(AbstractOperation): ...@@ -587,17 +549,9 @@ class SFG(AbstractOperation):
return self._precedence_list return self._precedence_list
def _traverse_for_precedence_list(self, first_iter_ports): def _traverse_for_precedence_list(self, first_iter_ports: List[OutputPort]) -> List[List[OutputPort]]:
# Find dependencies of output ports and input ports. # Find dependencies of output ports and input ports.
outports_per_inport = defaultdict(list) remaining_inports_per_operation = {op: op.input_count for op in self.operations}
remaining_inports_per_outport = dict()
for op in self.operations:
op_inputs = op.inputs
for out_i, outport in enumerate(op.outputs):
dependendent_indexes = op.inputs_required_for_output(out_i)
remaining_inports_per_outport[outport] = len(dependendent_indexes)
for in_i in dependendent_indexes:
outports_per_inport[op_inputs[in_i]].append(outport)
# Traverse output ports for precedence # Traverse output ports for precedence
curr_iter_ports = first_iter_ports curr_iter_ports = first_iter_ports
...@@ -614,11 +568,113 @@ class SFG(AbstractOperation): ...@@ -614,11 +568,113 @@ class SFG(AbstractOperation):
new_inport = signal.destination new_inport = signal.destination
# Don't traverse over Registers # Don't traverse over Registers
if new_inport is not None and not isinstance(new_inport.operation, Register): if new_inport is not None and not isinstance(new_inport.operation, Register):
for new_outport in outports_per_inport[new_inport]: new_op = new_inport.operation
remaining_inports_per_outport[new_outport] -= 1 remaining_inports_per_operation[new_op] -= 1
if remaining_inports_per_outport[new_outport] == 0: if remaining_inports_per_operation[new_op] == 0:
next_iter_ports.append(new_outport) next_iter_ports.extend(new_op.outputs)
curr_iter_ports = next_iter_ports curr_iter_ports = next_iter_ports
return precedence_list return precedence_list
def print_precedence_graph(self) -> None:
"""Prints a representation of the SFG's precedence list to the standard out.
If the precedence list already has been calculated then it uses the cached version,
otherwise it calculates the precedence list and then prints it."""
precedence_list = self.get_precedence_list()
line = "-" * 120
out_str = StringIO()
out_str.write(line)
printed_ops = set()
for iter_num, iter in enumerate(precedence_list, start=1):
for outport_num, outport in enumerate(iter, start=1):
if outport not in printed_ops:
# Only print once per operation, even if it has multiple outports
out_str.write("\n")
out_str.write(str(iter_num))
out_str.write(".")
out_str.write(str(outport_num))
out_str.write(" \t")
out_str.write(str(outport.operation))
printed_ops.add(outport)
out_str.write("\n")
out_str.write(line)
print(out_str.getvalue())
def get_operations_topological_order(self) -> Iterable[Operation]:
"""Returns an Iterable of the Operations in the SFG in Topological Order.
Feedback loops makes an absolutely correct Topological order impossible, so an
approximative Topological Order is returned in such cases in this implementation."""
if self._operations_topological_order:
return self._operations_topological_order
no_inputs_queue = deque(list(filter(lambda op: op.input_count == 0, self.operations)))
remaining_inports_per_operation = {op: op.input_count for op in self.operations}
# Maps number of input counts to a queue of seen objects with such a size.
seen_with_inputs_dict = defaultdict(deque)
seen = set()
top_order = []
assert len(no_inputs_queue) > 0, "Illegal SFG state, dangling signals in SFG."
first_op = no_inputs_queue.popleft()
visited = set([first_op])
p_queue = PriorityQueue()
p_queue.put((-first_op.output_count, first_op)) # Negative priority as max-heap popping is wanted
operations_left = len(self.operations) - 1
seen_but_not_visited_count = 0
while operations_left > 0:
while not p_queue.empty():
op = p_queue.get()[1]
operations_left -= 1
top_order.append(op)
visited.add(op)
for neighbor_op in op.subsequent_operations:
if neighbor_op not in visited:
remaining_inports_per_operation[neighbor_op] -= 1
remaining_inports = remaining_inports_per_operation[neighbor_op]
if remaining_inports == 0:
p_queue.put((-neighbor_op.output_count, neighbor_op))
elif remaining_inports > 0:
if neighbor_op in seen:
seen_with_inputs_dict[remaining_inports + 1].remove(neighbor_op)
else:
seen.add(neighbor_op)
seen_but_not_visited_count += 1
seen_with_inputs_dict[remaining_inports].append(neighbor_op)
# Check if have to fetch Operations from somewhere else since p_queue is empty
if operations_left > 0:
# First check if can fetch from Operations with no input ports
if no_inputs_queue:
new_op = no_inputs_queue.popleft()
p_queue.put((new_op.output_count, new_op))
# Else fetch operation with lowest input count that is not zero
elif seen_but_not_visited_count > 0:
for i in itertools.count(start=1):
seen_inputs_queue = seen_with_inputs_dict[i]
if seen_inputs_queue:
new_op = seen_inputs_queue.popleft()
p_queue.put((-new_op.output_count, new_op))
seen_but_not_visited_count -= 1
break
else:
raise RuntimeError("Unallowed structure in SFG detected")
self._operations_topological_order = top_order
return self._operations_topological_order
test/fixtures/signal_flow_graph.py
+ 90
− 26
View file @ 7b732af1
import pytest import pytest
from b_asic import SFG, Input, Output, Constant, Register, ConstantMultiplication from b_asic import SFG, Input, Output, Constant, Register, ConstantMultiplication, Addition, Butterfly
@pytest.fixture @pytest.fixture
def sfg_two_inputs_two_outputs(): def sfg_two_inputs_two_outputs():
"""Valid SFG with two inputs and two outputs. """Valid SFG with two inputs and two outputs.
. . . .
in1-------+ +--------->out1 in1-------+ +--------->out1
. | | . . | | .
. v | . . v | .
...@@ -17,9 +17,9 @@ def sfg_two_inputs_two_outputs(): ...@@ -17,9 +17,9 @@ def sfg_two_inputs_two_outputs():
| . ^ . | . ^ .
| . | . | . | .
+------------+ . +------------+ .
. . . .
out1 = in1 + in2 out1 = in1 + in2
out2 = in1 + 2 * in2 out2 = in1 + 2 * in2
""" """
in1 = Input() in1 = Input()
in2 = Input() in2 = Input()
...@@ -27,13 +27,14 @@ def sfg_two_inputs_two_outputs(): ...@@ -27,13 +27,14 @@ def sfg_two_inputs_two_outputs():
add2 = add1 + in2 add2 = add1 + in2
out1 = Output(add1) out1 = Output(add1)
out2 = Output(add2) out2 = Output(add2)
return SFG(inputs = [in1, in2], outputs = [out1, out2]) return SFG(inputs=[in1, in2], outputs=[out1, out2])
@pytest.fixture @pytest.fixture
def sfg_two_inputs_two_outputs_independent(): def sfg_two_inputs_two_outputs_independent():
"""Valid SFG with two inputs and two outputs, where the first output only depends """Valid SFG with two inputs and two outputs, where the first output only depends
on the first input and the second output only depends on the second input. on the first input and the second output only depends on the second input.
. . . .
in1-------------------->out1 in1-------------------->out1
. . . .
. . . .
...@@ -44,17 +45,18 @@ def sfg_two_inputs_two_outputs_independent(): ...@@ -44,17 +45,18 @@ def sfg_two_inputs_two_outputs_independent():
. | ^ . . | ^ .
. | | . . | | .
. +------+ . . +------+ .
. . . .
out1 = in1 out1 = in1
out2 = in2 + 3 out2 = in2 + 3
""" """
in1 = Input() in1 = Input("IN1")
in2 = Input() in2 = Input("IN2")
c1 = Constant(3) c1 = Constant(3, "C1")
add1 = in2 + c1 add1 = Addition(in2, c1, "ADD1")
out1 = Output(in1) out1 = Output(in1, "OUT1")
out2 = Output(add1) out2 = Output(add1, "OUT2")
return SFG(inputs = [in1, in2], outputs = [out1, out2]) return SFG(inputs=[in1, in2], outputs=[out1, out2])
@pytest.fixture @pytest.fixture
def sfg_nested(): def sfg_nested():
...@@ -65,7 +67,7 @@ def sfg_nested(): ...@@ -65,7 +67,7 @@ def sfg_nested():
mac_in2 = Input() mac_in2 = Input()
mac_in3 = Input() mac_in3 = Input()
mac_out1 = Output(mac_in1 + mac_in2 * mac_in3) mac_out1 = Output(mac_in1 + mac_in2 * mac_in3)
MAC = SFG(inputs = [mac_in1, mac_in2, mac_in3], outputs = [mac_out1]) MAC = SFG(inputs=[mac_in1, mac_in2, mac_in3], outputs=[mac_out1])
in1 = Input() in1 = Input()
in2 = Input() in2 = Input()
...@@ -73,7 +75,8 @@ def sfg_nested(): ...@@ -73,7 +75,8 @@ def sfg_nested():
mac2 = MAC(in1, in2, mac1) mac2 = MAC(in1, in2, mac1)
mac3 = MAC(in1, mac1, mac2) mac3 = MAC(in1, mac1, mac2)
out1 = Output(mac3) out1 = Output(mac3)
return SFG(inputs = [in1, in2], outputs = [out1]) return SFG(inputs=[in1, in2], outputs=[out1])
@pytest.fixture @pytest.fixture
def sfg_delay(): def sfg_delay():
...@@ -83,7 +86,8 @@ def sfg_delay(): ...@@ -83,7 +86,8 @@ def sfg_delay():
in1 = Input() in1 = Input()
reg1 = Register(in1) reg1 = Register(in1)
out1 = Output(reg1) out1 = Output(reg1)
return SFG(inputs = [in1], outputs = [out1]) return SFG(inputs=[in1], outputs=[out1])
@pytest.fixture @pytest.fixture
def sfg_accumulator(): def sfg_accumulator():
...@@ -95,7 +99,8 @@ def sfg_accumulator(): ...@@ -95,7 +99,8 @@ def sfg_accumulator():
reg = Register() reg = Register()
reg.input(0).connect((reg + data_in) * (1 - reset)) reg.input(0).connect((reg + data_in) * (1 - reset))
data_out = Output(reg) data_out = Output(reg)
return SFG(inputs = [data_in, reset], outputs = [data_out]) return SFG(inputs=[data_in, reset], outputs=[data_out])
@pytest.fixture @pytest.fixture
def simple_filter(): def simple_filter():
...@@ -105,11 +110,70 @@ def simple_filter(): ...@@ -105,11 +110,70 @@ def simple_filter():
| | | |
in1>------add1>------reg>------+------out1> in1>------add1>------reg>------+------out1>
""" """
in1 = Input() in1 = Input("IN1")
reg = Register() constmul1 = ConstantMultiplication(0.5, name="CMUL1")
constmul1 = ConstantMultiplication(0.5) add1 = Addition(in1, constmul1, "ADD1")
add1 = in1 + constmul1 reg = Register(add1, name="REG1")
reg.input(0).connect(add1)
constmul1.input(0).connect(reg) constmul1.input(0).connect(reg)
out1 = Output(reg) out1 = Output(reg, "OUT1")
return SFG(inputs=[in1], outputs=[out1]) return SFG(inputs=[in1], outputs=[out1], name="simple_filter")
@pytest.fixture
def precedence_sfg_registers():
"""A sfg with registers and interesting layout for precednce list generation.
IN1>--->C0>--->ADD1>--->Q1>---+--->A0>--->ADD4>--->OUT1
^ | ^
| T1 |
| | |
ADD2<---<B1<---+--->A1>--->ADD3
^ | ^
| T2 |
| | |
+-----<B2<---+--->A2>-----+
"""
in1 = Input("IN1")
c0 = ConstantMultiplication(5, in1, "C0")
add1 = Addition(c0, None, "ADD1")
# Not sure what operation "Q" is supposed to be in the example
Q1 = ConstantMultiplication(1, add1, "Q1")
T1 = Register(Q1, 0, "T1")
T2 = Register(T1, 0, "T2")
b2 = ConstantMultiplication(2, T2, "B2")
b1 = ConstantMultiplication(3, T1, "B1")
add2 = Addition(b1, b2, "ADD2")
add1.input(1).connect(add2)
a1 = ConstantMultiplication(4, T1, "A1")
a2 = ConstantMultiplication(6, T2, "A2")
add3 = Addition(a1, a2, "ADD3")
a0 = ConstantMultiplication(7, Q1, "A0")
add4 = Addition(a0, add3, "ADD4")
out1 = Output(add4, "OUT1")
return SFG(inputs=[in1], outputs=[out1], name="SFG")
@pytest.fixture
def precedence_sfg_registers_and_constants():
in1 = Input("IN1")
c0 = ConstantMultiplication(5, in1, "C0")
add1 = Addition(c0, None, "ADD1")
# Not sure what operation "Q" is supposed to be in the example
Q1 = ConstantMultiplication(1, add1, "Q1")
T1 = Register(Q1, 0, "T1")
const1 = Constant(10, "CONST1") # Replace T2 register with a constant
b2 = ConstantMultiplication(2, const1, "B2")
b1 = ConstantMultiplication(3, T1, "B1")
add2 = Addition(b1, b2, "ADD2")
add1.input(1).connect(add2)
a1 = ConstantMultiplication(4, T1, "A1")
a2 = ConstantMultiplication(10, const1, "A2")
add3 = Addition(a1, a2, "ADD3")
a0 = ConstantMultiplication(7, Q1, "A0")
# Replace ADD4 with a butterfly to test multiple output ports
bfly1 = Butterfly(a0, add3, "BFLY1")
out1 = Output(bfly1.output(0), "OUT1")
out2 = Output(bfly1.output(1), "OUT2")
return SFG(inputs=[in1], outputs=[out1], name="SFG")
test/test_sfg.py
+ 108
− 73
View file @ 7b732af1
import pytest import pytest
import io
import sys
from b_asic import SFG, Signal, Input, Output, Constant, ConstantMultiplication, Addition, Multiplication, Register, \ from b_asic import SFG, Signal, Input, Output, Constant, ConstantMultiplication, Addition, Multiplication, Register, \
Butterfly, Subtraction, SquareRoot Butterfly, Subtraction, SquareRoot
...@@ -54,13 +57,17 @@ class TestPrintSfg: ...@@ -54,13 +57,17 @@ class TestPrintSfg:
inp2 = Input("INP2") inp2 = Input("INP2")
add1 = Addition(inp1, inp2, "ADD1") add1 = Addition(inp1, inp2, "ADD1")
out1 = Output(add1, "OUT1") out1 = Output(add1, "OUT1")
sfg = SFG(inputs=[inp1, inp2], outputs=[out1], name="sf1") sfg = SFG(inputs=[inp1, inp2], outputs=[out1], name="SFG1")
assert sfg.__str__() == \ assert sfg.__str__() == \
"id: add1, name: ADD1, input: [s1, s2], output: [s3]\n" + \ "id: no_id, \tname: SFG1, \tinputs: {0: '-'}, \toutputs: {0: '-'}\n" + \
"id: in1, name: INP1, input: [], output: [s1]\n" + \ "Internal Operations:\n" + \
"id: in2, name: INP2, input: [], output: [s2]\n" + \ "----------------------------------------------------------------------------------------------------\n" + \
"id: out1, name: OUT1, input: [s3], output: []\n" str(sfg.find_by_name("INP1")[0]) + "\n" + \
str(sfg.find_by_name("INP2")[0]) + "\n" + \
str(sfg.find_by_name("ADD1")[0]) + "\n" + \
str(sfg.find_by_name("OUT1")[0]) + "\n" + \
"----------------------------------------------------------------------------------------------------\n"
def test_add_mul(self): def test_add_mul(self):
inp1 = Input("INP1") inp1 = Input("INP1")
...@@ -72,12 +79,16 @@ class TestPrintSfg: ...@@ -72,12 +79,16 @@ class TestPrintSfg:
sfg = SFG(inputs=[inp1, inp2, inp3], outputs=[out1], name="mac_sfg") sfg = SFG(inputs=[inp1, inp2, inp3], outputs=[out1], name="mac_sfg")
assert sfg.__str__() == \ assert sfg.__str__() == \
"id: add1, name: ADD1, input: [s1, s2], output: [s5]\n" + \ "id: no_id, \tname: mac_sfg, \tinputs: {0: '-'}, \toutputs: {0: '-'}\n" + \
"id: in1, name: INP1, input: [], output: [s1]\n" + \ "Internal Operations:\n" + \
"id: in2, name: INP2, input: [], output: [s2]\n" + \ "----------------------------------------------------------------------------------------------------\n" + \
"id: mul1, name: MUL1, input: [s5, s3], output: [s4]\n" + \ str(sfg.find_by_name("INP1")[0]) + "\n" + \
"id: in3, name: INP3, input: [], output: [s3]\n" + \ str(sfg.find_by_name("INP2")[0]) + "\n" + \
"id: out1, name: OUT1, input: [s4], output: []\n" str(sfg.find_by_name("ADD1")[0]) + "\n" + \
str(sfg.find_by_name("INP3")[0]) + "\n" + \
str(sfg.find_by_name("MUL1")[0]) + "\n" + \
str(sfg.find_by_name("OUT1")[0]) + "\n" + \
"----------------------------------------------------------------------------------------------------\n"
def test_constant(self): def test_constant(self):
inp1 = Input("INP1") inp1 = Input("INP1")
...@@ -88,18 +99,27 @@ class TestPrintSfg: ...@@ -88,18 +99,27 @@ class TestPrintSfg:
sfg = SFG(inputs=[inp1], outputs=[out1], name="sfg") sfg = SFG(inputs=[inp1], outputs=[out1], name="sfg")
assert sfg.__str__() == \ assert sfg.__str__() == \
"id: add1, name: ADD1, input: [s3, s1], output: [s2]\n" + \ "id: no_id, \tname: sfg, \tinputs: {0: '-'}, \toutputs: {0: '-'}\n" + \
"id: c1, name: CONST, value: 3, input: [], output: [s3]\n" + \ "Internal Operations:\n" + \
"id: in1, name: INP1, input: [], output: [s1]\n" + \ "----------------------------------------------------------------------------------------------------\n" + \
"id: out1, name: OUT1, input: [s2], output: []\n" str(sfg.find_by_name("CONST")[0]) + "\n" + \
str(sfg.find_by_name("INP1")[0]) + "\n" + \
str(sfg.find_by_name("ADD1")[0]) + "\n" + \
str(sfg.find_by_name("OUT1")[0]) + "\n" + \
"----------------------------------------------------------------------------------------------------\n"
def test_simple_filter(self, simple_filter): def test_simple_filter(self, simple_filter):
assert simple_filter.__str__() == \ assert simple_filter.__str__() == \
'id: add1, name: , input: [s1, s3], output: [s4]\n' + \ "id: no_id, \tname: simple_filter, \tinputs: {0: '-'}, \toutputs: {0: '-'}\n" + \
'id: in1, name: , input: [], output: [s1]\n' + \ "Internal Operations:\n" + \
'id: cmul1, name: , input: [s5], output: [s3]\n' + \ "----------------------------------------------------------------------------------------------------\n" + \
'id: reg1, name: , input: [s4], output: [s5, s2]\n' + \ str(simple_filter.find_by_name("IN1")[0]) + "\n" + \
'id: out1, name: , input: [s2], output: []\n' str(simple_filter.find_by_name("ADD1")[0]) + "\n" + \
str(simple_filter.find_by_name("REG1")[0]) + "\n" + \
str(simple_filter.find_by_name("CMUL1")[0]) + "\n" + \
str(simple_filter.find_by_name("OUT1")[0]) + "\n" + \
"----------------------------------------------------------------------------------------------------\n"
class TestDeepCopy: class TestDeepCopy:
...@@ -267,7 +287,7 @@ class TestInsertComponent: ...@@ -267,7 +287,7 @@ class TestInsertComponent:
_sfg = sfg.insert_operation(sqrt, sfg.find_by_name("constant4")[0].graph_id) _sfg = sfg.insert_operation(sqrt, sfg.find_by_name("constant4")[0].graph_id)
assert _sfg.evaluate() != sfg.evaluate() assert _sfg.evaluate() != sfg.evaluate()
assert any([isinstance(comp, SquareRoot) for comp in _sfg.operations]) assert any([isinstance(comp, SquareRoot) for comp in _sfg.operations])
assert not any([isinstance(comp, SquareRoot) for comp in sfg.operations]) assert not any([isinstance(comp, SquareRoot) for comp in sfg.operations])
...@@ -275,7 +295,8 @@ class TestInsertComponent: ...@@ -275,7 +295,8 @@ class TestInsertComponent:
assert isinstance(_sfg.find_by_name("constant4")[0].output(0).signals[0].destination.operation, SquareRoot) assert isinstance(_sfg.find_by_name("constant4")[0].output(0).signals[0].destination.operation, SquareRoot)
assert sfg.find_by_name("constant4")[0].output(0).signals[0].destination.operation is sfg.find_by_id("add3") assert sfg.find_by_name("constant4")[0].output(0).signals[0].destination.operation is sfg.find_by_id("add3")
assert _sfg.find_by_name("constant4")[0].output(0).signals[0].destination.operation is not _sfg.find_by_id("add3") assert _sfg.find_by_name("constant4")[0].output(
0).signals[0].destination.operation is not _sfg.find_by_id("add3")
assert _sfg.find_by_id("sqrt1").output(0).signals[0].destination.operation is _sfg.find_by_id("add3") assert _sfg.find_by_id("sqrt1").output(0).signals[0].destination.operation is _sfg.find_by_id("add3")
def test_insert_invalid_component_in_sfg(self, large_operation_tree): def test_insert_invalid_component_in_sfg(self, large_operation_tree):
...@@ -304,22 +325,26 @@ class TestInsertComponent: ...@@ -304,22 +325,26 @@ class TestInsertComponent:
assert len(_sfg.find_by_name("n_bfly")) == 1 assert len(_sfg.find_by_name("n_bfly")) == 1
# Correctly connected old output -> new input # Correctly connected old output -> new input
assert _sfg.find_by_name("bfly3")[0].output(0).signals[0].destination.operation is _sfg.find_by_name("n_bfly")[0] assert _sfg.find_by_name("bfly3")[0].output(
assert _sfg.find_by_name("bfly3")[0].output(1).signals[0].destination.operation is _sfg.find_by_name("n_bfly")[0] 0).signals[0].destination.operation is _sfg.find_by_name("n_bfly")[0]
assert _sfg.find_by_name("bfly3")[0].output(
1).signals[0].destination.operation is _sfg.find_by_name("n_bfly")[0]
# Correctly connected new input -> old output # Correctly connected new input -> old output
assert _sfg.find_by_name("n_bfly")[0].input(0).signals[0].source.operation is _sfg.find_by_name("bfly3")[0] assert _sfg.find_by_name("n_bfly")[0].input(0).signals[0].source.operation is _sfg.find_by_name("bfly3")[0]
assert _sfg.find_by_name("n_bfly")[0].input(1).signals[0].source.operation is _sfg.find_by_name("bfly3")[0] assert _sfg.find_by_name("n_bfly")[0].input(1).signals[0].source.operation is _sfg.find_by_name("bfly3")[0]
# Correctly connected new output -> next input # Correctly connected new output -> next input
assert _sfg.find_by_name("n_bfly")[0].output(0).signals[0].destination.operation is _sfg.find_by_name("bfly2")[0] assert _sfg.find_by_name("n_bfly")[0].output(
assert _sfg.find_by_name("n_bfly")[0].output(1).signals[0].destination.operation is _sfg.find_by_name("bfly2")[0] 0).signals[0].destination.operation is _sfg.find_by_name("bfly2")[0]
assert _sfg.find_by_name("n_bfly")[0].output(
1).signals[0].destination.operation is _sfg.find_by_name("bfly2")[0]
# Correctly connected next input -> new output # Correctly connected next input -> new output
assert _sfg.find_by_name("bfly2")[0].input(0).signals[0].source.operation is _sfg.find_by_name("n_bfly")[0] assert _sfg.find_by_name("bfly2")[0].input(0).signals[0].source.operation is _sfg.find_by_name("n_bfly")[0]
assert _sfg.find_by_name("bfly2")[0].input(1).signals[0].source.operation is _sfg.find_by_name("n_bfly")[0] assert _sfg.find_by_name("bfly2")[0].input(1).signals[0].source.operation is _sfg.find_by_name("n_bfly")[0]
class TestFindComponentsWithTypeName: class TestFindComponentsWithTypeName:
def test_mac_components(self): def test_mac_components(self):
inp1 = Input("INP1") inp1 = Input("INP1")
...@@ -358,28 +383,9 @@ class TestFindComponentsWithTypeName: ...@@ -358,28 +383,9 @@ class TestFindComponentsWithTypeName:
class TestGetPrecedenceList: class TestGetPrecedenceList:
def test_inputs_registers(self): def test_inputs_registers(self, precedence_sfg_registers):
in1 = Input("IN1")
c0 = ConstantMultiplication(5, in1, "C0")
add1 = Addition(c0, None, "ADD1")
# Not sure what operation "Q" is supposed to be in the example
Q1 = ConstantMultiplication(1, add1, "Q1")
T1 = Register(Q1, 0, "T1")
T2 = Register(T1, 0, "T2")
b2 = ConstantMultiplication(2, T2, "B2")
b1 = ConstantMultiplication(3, T1, "B1")
add2 = Addition(b1, b2, "ADD2")
add1.input(1).connect(add2)
a1 = ConstantMultiplication(4, T1, "A1")
a2 = ConstantMultiplication(6, T2, "A2")
add3 = Addition(a1, a2, "ADD3")
a0 = ConstantMultiplication(7, Q1, "A0")
add4 = Addition(a0, add3, "ADD4")
out1 = Output(add4, "OUT1")
sfg = SFG(inputs=[in1], outputs=[out1], name="SFG")
precedence_list = sfg.get_precedence_list() precedence_list = precedence_sfg_registers.get_precedence_list()
assert len(precedence_list) == 7 assert len(precedence_list) == 7
...@@ -404,30 +410,9 @@ class TestGetPrecedenceList: ...@@ -404,30 +410,9 @@ class TestGetPrecedenceList:
assert set([port.operation.key(port.index, port.operation.name) assert set([port.operation.key(port.index, port.operation.name)
for port in precedence_list[6]]) == {"ADD4"} for port in precedence_list[6]]) == {"ADD4"}
def test_inputs_constants_registers_multiple_outputs(self): def test_inputs_constants_registers_multiple_outputs(self, precedence_sfg_registers_and_constants):
in1 = Input("IN1")
c0 = ConstantMultiplication(5, in1, "C0")
add1 = Addition(c0, None, "ADD1")
# Not sure what operation "Q" is supposed to be in the example
Q1 = ConstantMultiplication(1, add1, "Q1")
T1 = Register(Q1, 0, "T1")
const1 = Constant(10, "CONST1") # Replace T2 register with a constant
b2 = ConstantMultiplication(2, const1, "B2")
b1 = ConstantMultiplication(3, T1, "B1")
add2 = Addition(b1, b2, "ADD2")
add1.input(1).connect(add2)
a1 = ConstantMultiplication(4, T1, "A1")
a2 = ConstantMultiplication(10, const1, "A2")
add3 = Addition(a1, a2, "ADD3")
a0 = ConstantMultiplication(7, Q1, "A0")
# Replace ADD4 with a butterfly to test multiple output ports
bfly1 = Butterfly(a0, add3, "BFLY1")
out1 = Output(bfly1.output(0), "OUT1")
out2 = Output(bfly1.output(1), "OUT2")
sfg = SFG(inputs=[in1], outputs=[out1], name="SFG")
precedence_list = sfg.get_precedence_list() precedence_list = precedence_sfg_registers_and_constants.get_precedence_list()
assert len(precedence_list) == 7 assert len(precedence_list) == 7
...@@ -502,10 +487,48 @@ class TestGetPrecedenceList: ...@@ -502,10 +487,48 @@ class TestGetPrecedenceList:
for port in precedence_list[0]]) == {"IN1", "IN2"} for port in precedence_list[0]]) == {"IN1", "IN2"}
assert set([port.operation.key(port.index, port.operation.name) assert set([port.operation.key(port.index, port.operation.name)
for port in precedence_list[1]]) == {"NESTED_SFG.0", "CMUL1"} for port in precedence_list[1]]) == {"CMUL1"}
assert set([port.operation.key(port.index, port.operation.name) assert set([port.operation.key(port.index, port.operation.name)
for port in precedence_list[2]]) == {"NESTED_SFG.1"} for port in precedence_list[2]]) == {"NESTED_SFG.0", "NESTED_SFG.1"}
class TestPrintPrecedence:
def test_registers(self, precedence_sfg_registers):
sfg = precedence_sfg_registers
captured_output = io.StringIO()
sys.stdout = captured_output
sfg.print_precedence_graph()
sys.stdout = sys.__stdout__
captured_output = captured_output.getvalue()
assert captured_output == \
"-" * 120 + "\n" + \
"1.1 \t" + str(sfg.find_by_name("IN1")[0]) + "\n" + \
"1.2 \t" + str(sfg.find_by_name("T1")[0]) + "\n" + \
"1.3 \t" + str(sfg.find_by_name("T2")[0]) + "\n" + \
"-" * 120 + "\n" + \
"2.1 \t" + str(sfg.find_by_name("C0")[0]) + "\n" + \
"2.2 \t" + str(sfg.find_by_name("A1")[0]) + "\n" + \
"2.3 \t" + str(sfg.find_by_name("B1")[0]) + "\n" + \
"2.4 \t" + str(sfg.find_by_name("A2")[0]) + "\n" + \
"2.5 \t" + str(sfg.find_by_name("B2")[0]) + "\n" + \
"-" * 120 + "\n" + \
"3.1 \t" + str(sfg.find_by_name("ADD3")[0]) + "\n" + \
"3.2 \t" + str(sfg.find_by_name("ADD2")[0]) + "\n" + \
"-" * 120 + "\n" + \
"4.1 \t" + str(sfg.find_by_name("ADD1")[0]) + "\n" + \
"-" * 120 + "\n" + \
"5.1 \t" + str(sfg.find_by_name("Q1")[0]) + "\n" + \
"-" * 120 + "\n" + \
"6.1 \t" + str(sfg.find_by_name("A0")[0]) + "\n" + \
"-" * 120 + "\n" + \
"7.1 \t" + str(sfg.find_by_name("ADD4")[0]) + "\n" + \
"-" * 120 + "\n"
class TestDepends: class TestDepends:
...@@ -672,3 +695,15 @@ class TestConnectExternalSignalsToComponentsMultipleComp: ...@@ -672,3 +695,15 @@ class TestConnectExternalSignalsToComponentsMultipleComp:
sfg1.connect_external_signals_to_components() sfg1.connect_external_signals_to_components()
assert test_sfg.evaluate(1, 2, 3, 4) == 16 assert test_sfg.evaluate(1, 2, 3, 4) == 16
assert not test_sfg.connect_external_signals_to_components() assert not test_sfg.connect_external_signals_to_components()
class TestTopologicalOrderOperations:
def test_feedback_sfg(self, simple_filter):
topological_order = simple_filter.get_operations_topological_order()
assert [comp.name for comp in topological_order] == ["IN1", "ADD1", "REG1", "CMUL1", "OUT1"]
def test_multiple_independent_inputs(self, sfg_two_inputs_two_outputs_independent):
topological_order = sfg_two_inputs_two_outputs_independent.get_operations_topological_order()
assert [comp.name for comp in topological_order] == ["IN1", "OUT1", "IN2", "C1", "ADD1", "OUT2"]
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment