Skip to content
Snippets Groups Projects
Commit 98e3b5c9 authored by Oscar Gustafsson's avatar Oscar Gustafsson :bicyclist:
Browse files

Refactor source/destination code of signal and operation

parent a23a48c3
No related branches found
No related tags found
1 merge request!133Refactor source/destination code of signal and operation
Pipeline #88545 passed
...@@ -389,6 +389,24 @@ class Operation(GraphComponent, SignalSourceProvider): ...@@ -389,6 +389,24 @@ class Operation(GraphComponent, SignalSourceProvider):
""" """
raise NotImplementedError raise NotImplementedError
@property
@abstractmethod
def source(self) -> OutputPort:
"""
Return the OutputPort if there is only one output port.
If not, raise a TypeError.
"""
raise NotImplementedError
@property
@abstractmethod
def destination(self) -> InputPort:
"""
Return the InputPort if there is only one input port.
If not, raise a TypeError.
"""
raise NotImplementedError
@abstractmethod @abstractmethod
def _increase_time_resolution(self, factor: int) -> None: def _increase_time_resolution(self, factor: int) -> None:
raise NotImplementedError raise NotImplementedError
...@@ -848,10 +866,20 @@ class AbstractOperation(Operation, AbstractGraphComponent): ...@@ -848,10 +866,20 @@ class AbstractOperation(Operation, AbstractGraphComponent):
diff = "more" if self.output_count > 1 else "less" diff = "more" if self.output_count > 1 else "less"
raise TypeError( raise TypeError(
f"{self.__class__.__name__} cannot be used as an input source" f"{self.__class__.__name__} cannot be used as an input source"
f" because it has {diff} than 1 output" f" because it has {diff} than one output"
) )
return self.output(0) return self.output(0)
@property
def destination(self) -> OutputPort:
if self.input_count != 1:
diff = "more" if self.input_count > 1 else "less"
raise TypeError(
f"{self.__class__.__name__} cannot be used as an output"
f" destination because it has {diff} than one input"
)
return self.input(0)
def truncate_input(self, index: int, value: Number, bits: int) -> Number: def truncate_input(self, index: int, value: Number, bits: int) -> Number:
return int(value) & ((2**bits) - 1) return int(value) & ((2**bits) - 1)
......
...@@ -13,8 +13,8 @@ from b_asic.graph_component import ( ...@@ -13,8 +13,8 @@ from b_asic.graph_component import (
) )
if TYPE_CHECKING: if TYPE_CHECKING:
from b_asic.port import InputPort, OutputPort
from b_asic.operation import Operation from b_asic.operation import Operation
from b_asic.port import InputPort, OutputPort
class Signal(AbstractGraphComponent): class Signal(AbstractGraphComponent):
...@@ -24,15 +24,19 @@ class Signal(AbstractGraphComponent): ...@@ -24,15 +24,19 @@ class Signal(AbstractGraphComponent):
Parameters Parameters
========== ==========
source : OutputPort or Operation, optional source : OutputPort, Signal, or Operation, optional
OutputPort or Operation to connect as source to the signal. OutputPort, Signal, or Operation to connect as source to the signal.
destination : InputPort or Operation, optional destination : InputPort, Signal, or Operation, optional
InputPort or Operation to connect as destination to the signal. InputPort, Signal, or Operation to connect as destination to the signal.
bits : int, optional bits : int, optional
The word length of the signal. The word length of the signal.
name : Name, default: "" name : Name, default: ""
The signal name. The signal name.
.. note:: If a Signal is provided as *source* or *destination*, the
connected port is used. Hence, if the argument signal is later
changed, it will not affect the current Signal.
See also See also
======== ========
set_source, set_destination set_source, set_destination
...@@ -43,8 +47,10 @@ class Signal(AbstractGraphComponent): ...@@ -43,8 +47,10 @@ class Signal(AbstractGraphComponent):
def __init__( def __init__(
self, self,
source: Optional[Union["OutputPort", "Operation"]] = None, source: Optional[Union["OutputPort", "Signal", "Operation"]] = None,
destination: Optional[Union["InputPort", "Operation"]] = None, destination: Optional[
Union["InputPort", "Signal", "Operation"]
] = None,
bits: Optional[int] = None, bits: Optional[int] = None,
name: Name = Name(""), name: Name = Name(""),
): ):
...@@ -80,7 +86,9 @@ class Signal(AbstractGraphComponent): ...@@ -80,7 +86,9 @@ class Signal(AbstractGraphComponent):
"""The destination InputPort of the signal.""" """The destination InputPort of the signal."""
return self._destination return self._destination
def set_source(self, source: Union["OutputPort", "Operation"]) -> None: def set_source(
self, source: Union["OutputPort", "Signal", "Operation"]
) -> None:
""" """
Disconnect the previous source OutputPort of the signal and Disconnect the previous source OutputPort of the signal and
connect to the entered source OutputPort. Also connect the entered connect to the entered source OutputPort. Also connect the entered
...@@ -89,21 +97,16 @@ class Signal(AbstractGraphComponent): ...@@ -89,21 +97,16 @@ class Signal(AbstractGraphComponent):
Parameters Parameters
========== ==========
source : OutputPort or Operation, optional source : OutputPort, Signal, or Operation, optional
OutputPort or Operation to connect as source to the signal. If OutputPort, Signal, or Operation to connect as source to the signal.
Operation, it must have a single output, otherwise a TypeError is If Signal, it will connect to the source of the signal, so later on
changing the source of the argument Signal will not affect this Signal.
If Operation, it must have a single output, otherwise a TypeError is
raised. That output is used to extract the OutputPort. raised. That output is used to extract the OutputPort.
""" """
from b_asic.operation import Operation if hasattr(source, "source"):
# Signal or Operation
if isinstance(source, Operation): source = source.source
if source.output_count != 1:
raise TypeError(
"Can only connect operations with a single output."
f" {source.type_name()} has {source.output_count} outputs."
" Use the output port directly instead."
)
source = source.output(0)
if source is not self._source: if source is not self._source:
self.remove_source() self.remove_source()
...@@ -111,7 +114,9 @@ class Signal(AbstractGraphComponent): ...@@ -111,7 +114,9 @@ class Signal(AbstractGraphComponent):
if self not in source.signals: if self not in source.signals:
source.add_signal(self) source.add_signal(self)
def set_destination(self, destination: "InputPort") -> None: def set_destination(
self, destination: Union["InputPort", "Signal", "Operation"]
) -> None:
""" """
Disconnect the previous destination InputPort of the signal and Disconnect the previous destination InputPort of the signal and
connect to the entered destination InputPort. Also connect the entered connect to the entered destination InputPort. Also connect the entered
...@@ -120,23 +125,17 @@ class Signal(AbstractGraphComponent): ...@@ -120,23 +125,17 @@ class Signal(AbstractGraphComponent):
Parameters Parameters
========== ==========
destination : InputPort or Operation destination : InputPort, Signal, or Operation
InputPort or Operation to connect as destination to the signal. InputPort, Signal, or Operation to connect as destination to the signal.
If Signal, it will connect to the destination of the signal, so later on
changing the destination of the argument Signal will not affect this Signal.
If Operation, it must have a single input, otherwise a TypeError If Operation, it must have a single input, otherwise a TypeError
is raised. is raised.
""" """
from b_asic.operation import Operation if hasattr(destination, "destination"):
# Signal or Operation
if isinstance(destination, Operation): destination = destination.destination
if destination.input_count != 1:
raise TypeError(
"Can only connect operations with a single input."
f" {destination.type_name()} has"
f" {destination.input_count} outputs. Use the input port"
" directly instead."
)
destination = destination.input(0)
if destination is not self._destination: if destination is not self._destination:
self.remove_destination() self.remove_destination()
......
...@@ -111,8 +111,8 @@ def test_signal_errors(): ...@@ -111,8 +111,8 @@ def test_signal_errors():
with pytest.raises( with pytest.raises(
TypeError, TypeError,
match=( match=(
"Can only connect operations with a single input. add has 2" "Addition cannot be used as an output destination because it has"
" outputs." " more than one input"
), ),
): ):
_ = Signal(cm1, add1) _ = Signal(cm1, add1)
...@@ -121,8 +121,8 @@ def test_signal_errors(): ...@@ -121,8 +121,8 @@ def test_signal_errors():
with pytest.raises( with pytest.raises(
TypeError, TypeError,
match=( match=(
"Can only connect operations with a single output. bfly has 2" "Butterfly cannot be used as an input source because it has more"
" outputs." " than one output"
), ),
): ):
_ = Signal(bf, cm1) _ = Signal(bf, cm1)
...@@ -132,8 +132,8 @@ def test_signal_errors(): ...@@ -132,8 +132,8 @@ def test_signal_errors():
with pytest.raises( with pytest.raises(
TypeError, TypeError,
match=( match=(
"Can only connect operations with a single input. add has 2" "Addition cannot be used as an output destination because it has"
" outputs." " more than one input"
), ),
): ):
signal.set_destination(add1) signal.set_destination(add1)
...@@ -141,8 +141,8 @@ def test_signal_errors(): ...@@ -141,8 +141,8 @@ def test_signal_errors():
with pytest.raises( with pytest.raises(
TypeError, TypeError,
match=( match=(
"Can only connect operations with a single output. bfly has 2" "Butterfly cannot be used as an input source because it has more"
" outputs." " than one output"
), ),
): ):
signal.set_source(bf) signal.set_source(bf)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment