Skip to content
Snippets Groups Projects

Resolve "Replacing Operations in SFG"

Merged Jacob Wahlman requested to merge 15-replacing-operations-in-sfg into develop
All threads resolved!
2 files
+ 35
56
Compare changes
  • Side-by-side
  • Inline
Files
2
+ 24
34
@@ -5,7 +5,7 @@ TODO: More info.
@@ -5,7 +5,7 @@ TODO: More info.
from typing import NewType, List, Iterable, Sequence, Dict, Optional, DefaultDict, Set
from typing import NewType, List, Iterable, Sequence, Dict, Optional, DefaultDict, Set
from numbers import Number
from numbers import Number
from collections import defaultdict, deque, Iterable as CIterable
from collections import defaultdict, deque
from b_asic.port import SignalSourceProvider, OutputPort
from b_asic.port import SignalSourceProvider, OutputPort
from b_asic.operation import Operation, AbstractOperation
from b_asic.operation import Operation, AbstractOperation
@@ -350,50 +350,40 @@ class SFG(AbstractOperation):
@@ -350,50 +350,40 @@ class SFG(AbstractOperation):
input_values.append(self._evaluate_source(input_src))
input_values.append(self._evaluate_source(input_src))
return src.operation.evaluate_output(src.index, input_values)
return src.operation.evaluate_output(src.index, input_values)
def replace_component(self, _component: List[Operation], _id: List[GraphID] = None, _type: List[Operation] = None):
def replace_component(self, component: Operation, _component: Operation = None, _id: GraphID = None):
"""Find and replace all components matching either on GraphID, Type or both.
"""Find and replace all components matching either on GraphID, Type or both.
Then return a new deepcopy of the sfg with the replaced component.
Then return a new deepcopy of the sfg with the replaced component.
Arguments:
Arguments:
_component: The list of new component(s), e.g Multiplication
component: The new component(s), e.g Multiplication
Keyword arguments:
Keyword arguments:
_id: The list of GraphID(s) to match the component to replace.
_component: The specific component to replace.
_type: The list of Type(s) to match the component to replace.
_id: The GraphID to match the component to replace.
"""
"""
_id = [_id] if not isinstance(_id, CIterable) or isinstance(_id, str) else _id
assert _component is not None or _id is not None, \
_type = [_type] if not isinstance(_type, CIterable) else _type
"Define either operation to replace or GraphID of operation"
_component = [_component] if not isinstance(_component, CIterable) else _component
components = set()
for comp_id in _id:
if comp_id is None:
continue
components |= {value for key, value in self._components_by_id.items() if key == comp_id}
if _id is not None:
 
_component = self.find_by_id(_id)
for comp_type in _type:
assert _component is not None and isinstance(_component, Operation), \
if comp_type is None:
"No operation matching the criteria found"
continue
assert _component.output_count == component.output_count, \
 
"The output count may not differ between the operations"
 
assert _component.input_count == component.input_count, \
 
"The input count may not differ between the operations"
components |= {comp for comp in self.components if isinstance(comp, comp_type)}
for index_in, _inp in enumerate(_component.inputs):
 
for _signal in _inp.signals:
 
_signal.remove_destination()
 
_signal.set_destination(component.input(index_in))
assert sum([comp.output_count for comp in _component]) == sum([comp.output_count for comp in components])
for index_out, _out in enumerate(_component.outputs):
assert sum([comp.input_count for comp in _component]) == sum([comp.input_count for comp in components])
for _signal in _out.signals:
_signal.remove_source()
for index, comp in enumerate(components):
_signal.set_source(component.output(index_out))
component = _component if not isinstance(_component, CIterable) else _component[index]
for index_in, _inp in enumerate(comp.inputs):
for _signal in _inp.signals:
_signal.remove_destination()
_signal.set_destination(component.input(index_in))
for index_out, _out in enumerate(comp.outputs):
for _signal in _out.signals:
_signal.remove_source()
_signal.set_source(component.output(index_out))
# The old SFG will be deleted by Python GC
# The old SFG will be deleted by Python GC
return self.deep_copy()
return self.deep_copy()
Loading