From c4b4fdaeba80f1f15ff80bf76adbb8ef00861d89 Mon Sep 17 00:00:00 2001
From: Angus Lothian <anglo547@student.liu.se>
Date: Thu, 12 Mar 2020 12:12:08 +0100
Subject: [PATCH] Add updated port and signal interface where connecting /
disconnecting is done reflectibly between both signal and port, also add some
more helper methods.
---
b_asic/abstract_graph_component.py | 1 +
b_asic/abstract_operation.py | 11 +-
b_asic/core_operations.py | 35 ++++---
b_asic/graph_component.py | 6 +-
b_asic/graph_id.py | 2 +-
b_asic/operation.py | 2 -
b_asic/port.py | 156 ++++++++++++++++++++--------
b_asic/signal.py | 81 +++++++++++----
b_asic/signal_flow_graph.py | 12 +--
test/fixtures/operation_tree.py | 16 +--
test/graph_id/conftest.py | 1 -
test/port/test_inputport.py | 84 +++++++++++++--
test/port/test_outputport.py | 35 +++++--
test/signal/test_signal.py | 62 +++++++++++
test/signal_flow_graph/conftest.py | 1 -
test/traverse/test_traverse_tree.py | 5 +-
16 files changed, 385 insertions(+), 125 deletions(-)
delete mode 100644 test/graph_id/conftest.py
create mode 100644 test/signal/test_signal.py
delete mode 100644 test/signal_flow_graph/conftest.py
diff --git a/b_asic/abstract_graph_component.py b/b_asic/abstract_graph_component.py
index 6efb94e3..a0b71b41 100644
--- a/b_asic/abstract_graph_component.py
+++ b/b_asic/abstract_graph_component.py
@@ -5,6 +5,7 @@ TODO: More info.
from b_asic.graph_component import GraphComponent, Name
+
class AbstractGraphComponent(GraphComponent):
"""Abstract Graph Component class which is a component of a signal flow graph.
diff --git a/b_asic/abstract_operation.py b/b_asic/abstract_operation.py
index 7cee1790..1403f7a9 100644
--- a/b_asic/abstract_operation.py
+++ b/b_asic/abstract_operation.py
@@ -4,7 +4,7 @@ TODO: More info.
"""
from abc import abstractmethod
-from typing import List, Set, Dict, Optional, Any
+from typing import List, Dict, Optional, Any
from numbers import Number
from b_asic.port import InputPort, OutputPort
@@ -13,6 +13,8 @@ from b_asic.operation import Operation
from b_asic.simulation import SimulationState, OperationState
from b_asic.utilities import breadth_first_search
from b_asic.abstract_graph_component import AbstractGraphComponent
+from b_asic.graph_component import Name
+
class AbstractOperation(Operation, AbstractGraphComponent):
"""Generic abstract operation class which most implementations will derive from.
@@ -23,15 +25,16 @@ class AbstractOperation(Operation, AbstractGraphComponent):
_output_ports: List[OutputPort]
_parameters: Dict[str, Optional[Any]]
- def __init__(self, **kwds):
- super().__init__(**kwds)
+ def __init__(self, name: Name = ""):
+ super().__init__(name)
self._input_ports = []
self._output_ports = []
self._parameters = {}
@abstractmethod
def evaluate(self, inputs: list) -> list:
- """Evaluate the operation and generate a list of output values given a list of input values."""
+ """Evaluate the operation and generate a list of output values given a
+ list of input values."""
raise NotImplementedError
def inputs(self) -> List[InputPort]:
diff --git a/b_asic/core_operations.py b/b_asic/core_operations.py
index f023e1a5..42867aa5 100644
--- a/b_asic/core_operations.py
+++ b/b_asic/core_operations.py
@@ -29,10 +29,10 @@ class Constant(AbstractOperation):
TODO: More info.
"""
- def __init__(self, value: Number, **kwds):
- """Construct a Constant."""
- super().__init__(**kwds)
- self._output_ports = [OutputPort(1, self)] # TODO: Generate appropriate ID for ports.
+ def __init__(self, value: Number = 0, name: Name = ""):
+ super().__init__(name)
+
+ self._output_ports = [OutputPort(0, self)] # TODO: Generate appropriate ID for ports.
self._parameters["value"] = value
def evaluate(self, inputs: list) -> list:
@@ -48,11 +48,16 @@ class Addition(AbstractOperation):
TODO: More info.
"""
- def __init__(self, **kwds):
- """Construct an Addition."""
- super().__init__(**kwds)
- self._input_ports = [InputPort(1, self), InputPort(1, self)] # TODO: Generate appropriate ID for ports.
- self._output_ports = [OutputPort(1, self)] # TODO: Generate appropriate ID for ports.
+ def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
+ super().__init__(name)
+
+ self._input_ports = [InputPort(0, self), InputPort(1, self)] # TODO: Generate appropriate ID for ports.
+ self._output_ports = [OutputPort(0, self)] # TODO: Generate appropriate ID for ports.
+
+ if source1 is not None:
+ self._input_ports[0].connect_to_port(source1)
+ if source2 is not None:
+ self._input_ports[1].connect_to_port(source2)
def evaluate(self, inputs: list) -> list:
return [inputs[0] + inputs[1]]
@@ -67,13 +72,15 @@ class ConstantMultiplication(AbstractOperation):
TODO: More info.
"""
- def __init__(self, coefficient: Number, **kwds):
- """Construct a ConstantMultiplication."""
- super().__init__(**kwds)
- self._input_ports = [InputPort(1), self] # TODO: Generate appropriate ID for ports.
- self._output_ports = [OutputPort(1, self)] # TODO: Generate appropriate ID for ports.
+ def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
+ super().__init__(name)
+ self._input_ports = [InputPort(0, self)] # TODO: Generate appropriate ID for ports.
+ self._output_ports = [OutputPort(0, self)] # TODO: Generate appropriate ID for ports.
self._parameters["coefficient"] = coefficient
+ if source1 is not None:
+ self._input_ports[0].connect_to_port(source1)
+
def evaluate(self, inputs: list) -> list:
return [inputs[0] * self.param("coefficient")]
diff --git a/b_asic/graph_component.py b/b_asic/graph_component.py
index 4bce63f2..ec39a28d 100644
--- a/b_asic/graph_component.py
+++ b/b_asic/graph_component.py
@@ -18,17 +18,17 @@ class GraphComponent(ABC):
@property
@abstractmethod
def type_name(self) -> TypeName:
- """Returns the type name of the graph component"""
+ """Return the type name of the graph component"""
raise NotImplementedError
@property
@abstractmethod
def name(self) -> Name:
- """Returns the name of the graph component."""
+ """Return the name of the graph component."""
raise NotImplementedError
@name.setter
@abstractmethod
def name(self, name: Name) -> None:
- """Sets the name of the graph component to the entered name."""
+ """Set the name of the graph component to the entered name."""
raise NotImplementedError
diff --git a/b_asic/graph_id.py b/b_asic/graph_id.py
index 0fd1855b..8da6a9d4 100644
--- a/b_asic/graph_id.py
+++ b/b_asic/graph_id.py
@@ -20,7 +20,7 @@ class GraphIDGenerator:
self._next_id_number = defaultdict(lambda: 1) # Initalises every key element to 1
def get_next_id(self, graph_id_type: GraphIDType) -> GraphID:
- """Returns the next graph id for a certain graph id type."""
+ """Return the next graph id for a certain graph id type."""
graph_id = graph_id_type + str(self._next_id_number[graph_id_type])
self._next_id_number[graph_id_type] += 1 # Increase the current id number
return graph_id
diff --git a/b_asic/operation.py b/b_asic/operation.py
index 4a716b79..acd26672 100644
--- a/b_asic/operation.py
+++ b/b_asic/operation.py
@@ -43,13 +43,11 @@ class Operation(GraphComponent):
"""Get the input port at index i."""
raise NotImplementedError
-
@abstractmethod
def output(self, i: int) -> "OutputPort":
"""Get the output port at index i."""
raise NotImplementedError
-
@abstractmethod
def params(self) -> Dict[str, Optional[Any]]:
"""Get a dictionary of all parameter values."""
diff --git a/b_asic/port.py b/b_asic/port.py
index 6cbd59ba..eff9db9d 100644
--- a/b_asic/port.py
+++ b/b_asic/port.py
@@ -6,15 +6,16 @@ TODO: More info.
from abc import ABC, abstractmethod
from typing import NewType, Optional, List
-from b_asic.signal import Signal
from b_asic.operation import Operation
+from b_asic.signal import Signal
PortId = NewType("PortId", int)
class Port(ABC):
"""Abstract port class.
- TODO: More info.
+
+ Handles functionality for port id and saves the connection to the parent operation.
"""
_port_id: PortId
@@ -25,39 +26,62 @@ class Port(ABC):
self._operation = operation
@property
- def identifier(self) -> PortId:
- """Get the unique identifier."""
+ def id(self) -> PortId:
+ """Return the unique portid."""
return self._port_id
@property
def operation(self) -> Operation:
- """Get the connected operation."""
+ """Return the connected operation."""
return self._operation
@property
@abstractmethod
def signals(self) -> List[Signal]:
- """Get a list of all connected signals."""
+ """Return a list of all connected signals."""
raise NotImplementedError
@abstractmethod
def signal(self, i: int = 0) -> Signal:
- """Get the connected signal at index i."""
+ """Return the connected signal at index i.
+
+ Keyword argumens:
+ i: integer index of the signal requsted.
+ """
+ raise NotImplementedError
+
+ @property
+ @abstractmethod
+ def connected_ports(self) -> List["Port"]:
+ """Return a list of all connected Ports."""
raise NotImplementedError
@abstractmethod
def signal_count(self) -> int:
- """Get the number of connected signals."""
+ """Return the number of connected signals."""
+ raise NotImplementedError
+
+ @abstractmethod
+ def connect_port(self, port: "Port") -> Signal:
+ """Create and return a signal that is connected to this port and the entered
+ port and connect this port to the signal and the entered port to the signal."""
raise NotImplementedError
@abstractmethod
- def connect(self, signal: Signal) -> None:
- """Connect a signal."""
+ def connect_signal(self, signal: Signal) -> None:
+ """Connect this port to the entered signal. If the entered signal isn't connected to
+ this port then connect the entered signal to the port aswell."""
raise NotImplementedError
@abstractmethod
- def disconnect(self, i: int = 0) -> None:
- """Disconnect a signal."""
+ def disconnect_signal(self, i: int = 0) -> None:
+ """Disconnect a signal from the port. If the port is still connected to the entered signal
+ then the port is disconnected from the the entered signal aswell."""
+ raise NotImplementedError
+
+ @abstractmethod
+ def is_connected_to_signal(self, signal: Signal) -> bool:
+ """Return true if the port is connected to the entered signal else false."""
raise NotImplementedError
@@ -65,34 +89,51 @@ class InputPort(Port):
"""Input port.
TODO: More info.
"""
- _source_signal: Optional[Signal]
+
+ _signal: Optional[Signal]
def __init__(self, port_id: PortId, operation: Operation):
super().__init__(port_id, operation)
- self._source_signal = None
+ self._signal = None
@property
def signals(self) -> List[Signal]:
- return [] if self._source_signal is None else [self._source_signal]
+ return [] if self._signal is None else [self._signal]
def signal(self, i: int = 0) -> Signal:
- assert 0 <= i < self.signal_count() # TODO: Error message.
- assert self._source_signal is not None # TODO: Error message.
- return self._source_signal
+ assert 0 <= i < self.signal_count(), "Signal index out of bound."
+ assert self._signal is not None, "No Signal connect to InputPort."
+ return self._signal
+
+ @property
+ def connected_ports(self) -> List[Port]:
+ return [] if self._signal is None else [self._signal.source]
def signal_count(self) -> int:
- return 0 if self._source_signal is None else 1
+ return 0 if self._signal is None else 1
- def connect(self, signal: Signal) -> None:
- self._source_signal = signal
- signal.destination = self
+ def connect_port(self, port: "OutputPort") -> Signal:
+ assert self._signal is None, "Connecting new port to already connected input port."
+ return Signal(port, self) # self._signal is set by the signal constructor
- def disconnect(self, i: int = 0) -> None:
- assert 0 <= i < self.signal_count() # TODO: Error message.
- self._source_signal.disconnect_source()
- self._source_signal = None
+ def connect_signal(self, signal: Signal) -> None:
+ assert self._signal is None, "Connecting new port to already connected input port."
+ self._signal = signal
+ if self is not signal.destination:
+ # Connect this inputport as destination for this signal if it isn't already.
+ signal.connect_destination(self)
- # TODO: More stuff.
+ def disconnect_signal(self, i: int = 0) -> None:
+ assert 0 <= i < self.signal_count(), "Signal Index out of range."
+ old_signal: Signal = self._signal
+ self._signal = None
+ if self is old_signal.destination:
+ # Disconnect the dest of the signal if this inputport currently is the dest
+ old_signal.disconnect_destination()
+ old_signal.disconnect_destination()
+
+ def is_connected_to_signal(self, signal: Signal) -> bool:
+ return self._signal is signal
class OutputPort(Port):
@@ -100,30 +141,57 @@ class OutputPort(Port):
TODO: More info.
"""
- _destination_signals: List[Signal]
+ _signals: List[Signal]
def __init__(self, port_id: PortId, operation: Operation):
super().__init__(port_id, operation)
- self._destination_signals = []
+ self._signals = []
@property
def signals(self) -> List[Signal]:
- return self._destination_signals.copy()
+ return self._signals.copy()
def signal(self, i: int = 0) -> Signal:
- assert 0 <= i < self.signal_count() # TODO: Error message.
- return self._destination_signals[i]
-
- def signal_count(self) -> int:
- return len(self._destination_signals)
+ assert 0 <= i < self.signal_count(), "Signal index out of bounds."
+ return self._signals[i]
- def connect(self, signal: Signal) -> None:
- assert signal not in self._destination_signals # TODO: Error message.
- self._destination_signals.append(signal)
- signal.source = self
-
- def disconnect(self, i: int = 0) -> None:
- assert 0 <= i < self.signal_count() # TODO: Error message.
- del self._destination_signals[i]
+ @property
+ def connected_ports(self) -> List[Port]:
+ return [signal.destination for signal in self._signals \
+ if signal.destination is not None]
- # TODO: More stuff.
+ def signal_count(self) -> int:
+ return len(self._signals)
+
+ def connect_port(self, port: InputPort) -> Signal:
+ return Signal(self, port) # Signal is added to self._signals in signal constructor
+
+ def connect_signal(self, signal: Signal) -> None:
+ assert not self.is_connected_to_signal(signal), \
+ "Attempting to connect to Signal already connected."
+ self._signals.append(signal)
+ if self is not signal.source:
+ # Connect this outputport to the signal if it isn't already
+ signal.connect_source(self)
+
+ def disconnect_signal(self, i: int = 0) -> None:
+ assert 0 <= i < self.signal_count(), "Signal index out of bounds."
+ old_signal: Signal = self._signals[i]
+ del self._signals[i]
+ if self is old_signal.source:
+ # Disconnect the source of the signal if this outputport currently is the source
+ old_signal.disconnect_source()
+
+ def disconnect_signal_by_ref(self, signal: Signal) -> None:
+ """Remove the signal that was entered from the OutputPorts signals.
+ If the entered signal still is connected to this port then disconnect the
+ entered signal from the port aswell.
+
+ Keyword arguments:
+ - signal: Signal to remove.
+ """
+ i: int = self._signals.index(signal)
+ self.disconnect_signal(i)
+
+ def is_connected_to_signal(self, signal: Signal) -> bool:
+ return signal in self._signals # O(n) complexity
diff --git a/b_asic/signal.py b/b_asic/signal.py
index 4d80530e..917e4af3 100644
--- a/b_asic/signal.py
+++ b/b_asic/signal.py
@@ -1,52 +1,97 @@
"""@package docstring
B-ASIC Signal Module.
"""
-from typing import TYPE_CHECKING, Optional
+from typing import Optional, TYPE_CHECKING
from b_asic.graph_component import TypeName
from b_asic.abstract_graph_component import AbstractGraphComponent
+from b_asic.graph_component import Name
if TYPE_CHECKING:
- from b_asic import OutputPort, InputPort
+ from b_asic.port import InputPort, OutputPort
class Signal(AbstractGraphComponent):
"""A connection between two ports."""
+
_source: "OutputPort"
_destination: "InputPort"
- def __init__(self, src: Optional["OutputPort"] = None, dest: Optional["InputPort"] = None, **kwds):
- super().__init__(**kwds)
- self._source = src
- self._destination = dest
+ def __init__(self, source: Optional["OutputPort"] = None, \
+ destination: Optional["InputPort"] = None, name: Name = ""):
+
+ super().__init__(name)
+
+ self._source = source
+ self._destination = destination
+
+ if source is not None:
+ self.connect_source(source)
+
+ if destination is not None:
+ self.connect_destination(destination)
@property
def source(self) -> "OutputPort":
- """Returns the source OutputPort of the signal."""
+ """Return the source OutputPort of the signal."""
return self._source
@property
def destination(self) -> "InputPort":
- """Returns the destination InputPort of the signal."""
+ """Return the destination "InputPort" of the signal."""
return self._destination
- @source.setter
- def source(self, src: "OutputPort") -> None:
- """Sets the value of the source OutputPort of the signal."""
+ def connect_source(self, src: "OutputPort") -> None:
+ """Disconnect the previous source OutputPort of the signal and
+ connect to the entered source OutputPort. Also connect the entered
+ source port to the signal if it hasn't already been connected.
+
+ Keyword arguments:
+ - src: OutputPort to connect as source to the signal.
+ """
+ self.disconnect_source()
self._source = src
+ if not src.is_connected_to_signal(self):
+ # If the new source isn't connected to this signal then connect it.
+ src.connect_signal(self)
- @destination.setter
- def destination(self, dest: "InputPort") -> None:
- """Sets the value of the destination InputPort of the signal."""
+ def connect_destination(self, dest: "InputPort") -> None:
+ """Disconnect the previous destination InputPort of the signal and
+ connect to the entered destination InputPort. Also connect the entered
+ destination port to the signal if it hasn't already been connected.
+
+ Keywords argments:
+ - dest: InputPort to connect as destination to the signal.
+ """
+ self.disconnect_destination()
self._destination = dest
+ if not dest.is_connected_to_signal(self):
+ # If the new destination isn't connected to tis signal then connect it.
+ dest.connect_signal(self)
@property
def type_name(self) -> TypeName:
return "s"
def disconnect_source(self) -> None:
- """Disconnects the source OutputPort of the signal."""
- self._source = None
+ """Disconnect the source OutputPort of the signal. If the source port
+ still is connected to this signal then also disconnect the source port."""
+ if self._source is not None:
+ old_source: "OutputPort" = self._source
+ self._source = None
+ if old_source.is_connected_to_signal(self):
+ # If the old destination port still is connected to this signal, then disconnect it.
+ old_source.disconnect_signal_by_ref(self)
def disconnect_destination(self) -> None:
- """Disconnects the destination InputPort of the signal."""
- self._destination = None
+ """Disconnect the destination InputPort of the signal."""
+ if self._destination is not None:
+ old_destination: "InputPort" = self._destination
+ self._destination = None
+ if old_destination.is_connected_to_signal(self):
+ # If the old destination port still is connected to this signal, then disconnect it.
+ old_destination.disconnect_signal()
+
+ def is_connected(self) -> bool:
+ """Returns true if the signal is connected to both a source and a destination,
+ else false."""
+ return self._source is not None and self._destination is not None
diff --git a/b_asic/signal_flow_graph.py b/b_asic/signal_flow_graph.py
index 6a91521a..ab2c3e94 100644
--- a/b_asic/signal_flow_graph.py
+++ b/b_asic/signal_flow_graph.py
@@ -50,8 +50,8 @@ class SFG(AbstractOperation):
return [] # TODO: Implement
def _add_graph_component(self, graph_component: GraphComponent) -> GraphID:
- """Adds the entered graph component to the SFG's dictionary of graph objects and
- returns a generated GraphID for it.
+ """Add the entered graph component to the SFG's dictionary of graph objects and
+ return a generated GraphID for it.
Keyword arguments:
graph_component: Graph component to add to the graph.
@@ -65,8 +65,8 @@ class SFG(AbstractOperation):
return graph_id
def find_by_id(self, graph_id: GraphID) -> Optional[GraphComponent]:
- """Finds a graph object based on the entered Graph ID and returns it. If no graph
- object with the entered ID was found then returns None.
+ """Find a graph object based on the entered Graph ID and return it. If no graph
+ object with the entered ID was found then return None.
Keyword arguments:
graph_id: Graph ID of the wanted object.
@@ -77,8 +77,8 @@ class SFG(AbstractOperation):
return None
def find_by_name(self, name: Name) -> List[GraphComponent]:
- """Finds all graph objects that have the entered name and returns them
- in a list. If no graph object with the entered name was found then returns an
+ """Find all graph objects that have the entered name and return them
+ in a list. If no graph object with the entered name was found then return an
empty list.
Keyword arguments:
diff --git a/test/fixtures/operation_tree.py b/test/fixtures/operation_tree.py
index 7cc9e422..b97e89d7 100644
--- a/test/fixtures/operation_tree.py
+++ b/test/fixtures/operation_tree.py
@@ -10,9 +10,9 @@ def operation():
def create_operation(_type, dest_oper, index, **kwargs):
oper = _type(**kwargs)
oper_signal = Signal()
- oper._output_ports[0].connect(oper_signal)
+ oper._output_ports[0].connect_signal(oper_signal)
- dest_oper._input_ports[index].connect(oper_signal)
+ dest_oper._input_ports[index].connect_signal(oper_signal)
return oper
@pytest.fixture
@@ -50,11 +50,11 @@ def large_operation_tree():
create_operation(Constant, add_oper_2, 1, value=5)
add_oper_3 = Addition()
- add_oper_signal = Signal(add_oper, add_oper_3)
- add_oper._output_ports[0].connect(add_oper_signal)
- add_oper_3._input_ports[0].connect(add_oper_signal)
+ add_oper_signal = Signal(add_oper.output(0), add_oper_3.output(0))
+ add_oper._output_ports[0].connect_signal(add_oper_signal)
+ add_oper_3._input_ports[0].connect_signal(add_oper_signal)
- add_oper_2_signal = Signal(add_oper_2, add_oper_3)
- add_oper_2._output_ports[0].connect(add_oper_2_signal)
- add_oper_3._input_ports[1].connect(add_oper_2_signal)
+ add_oper_2_signal = Signal(add_oper_2.output(0), add_oper_3.output(0))
+ add_oper_2._output_ports[0].connect_signal(add_oper_2_signal)
+ add_oper_3._input_ports[1].connect_signal(add_oper_2_signal)
return const_oper
diff --git a/test/graph_id/conftest.py b/test/graph_id/conftest.py
deleted file mode 100644
index 5871ed8e..00000000
--- a/test/graph_id/conftest.py
+++ /dev/null
@@ -1 +0,0 @@
-import pytest
diff --git a/test/port/test_inputport.py b/test/port/test_inputport.py
index f0e70cb7..7a78d5f7 100644
--- a/test/port/test_inputport.py
+++ b/test/port/test_inputport.py
@@ -2,19 +2,83 @@
B-ASIC test suite for Inputport
"""
-from b_asic import InputPort
-
import pytest
-def test_connect_multiple_signals(signals):
- """
- test if only one signal can connect to an input port
- """
+from b_asic import InputPort, OutputPort
+from b_asic import Signal
+
+@pytest.fixture
+def inp_port():
+ return InputPort(0, None)
+
+@pytest.fixture
+def out_port():
+ return OutputPort(0, None)
+
+@pytest.fixture
+def out_port2():
+ return OutputPort(1, None)
+
+@pytest.fixture
+def dangling_sig():
+ return Signal()
+
+@pytest.fixture
+def s_w_source():
+ out_port = OutputPort(0, None)
+ return Signal(source=out_port)
+
+@pytest.fixture
+def sig_with_dest():
+ inp_port = InputPort(0, None)
+ return Signal(destination=out_port)
+
+@pytest.fixture
+def connected_sig():
+ out_port = OutputPort(0, None)
inp_port = InputPort(0, None)
+ return Signal(source=out_port, destination=inp_port)
+
+def test_connect_port_then_disconnect(inp_port, out_port):
+ """Test connect unused port to port."""
+ s1 = inp_port.connect_port(out_port)
+
+ assert inp_port.connected_ports == [out_port]
+ assert out_port.connected_ports == [inp_port]
+ assert inp_port.signals == [s1]
+ assert out_port.signals == [s1]
+ assert s1.source is out_port
+ assert s1.destination is inp_port
+
+ inp_port.disconnect_signal()
+
+ assert inp_port.connected_ports == []
+ assert out_port.connected_ports == []
+ assert inp_port.signals == []
+ assert out_port.signals == [s1]
+ assert s1.source is out_port
+ assert s1.destination is None
+
+def test_connect_used_port_to_new_port(inp_port, out_port, out_port2):
+ """Does connecting multiple ports to an inputport throw error?"""
+ inp_port.connect_port(out_port)
+ with pytest.raises(AssertionError):
+ inp_port.connect_port(out_port2)
+
+def test_connect_signal_then_disconnect(inp_port, s_w_source):
+ inp_port.connect_signal(s_w_source)
+
+ assert inp_port.connected_ports == [s_w_source.source]
+ assert s_w_source.source.connected_ports == [inp_port]
+ assert inp_port.signals == [s_w_source]
+ assert s_w_source.source.signals == [s_w_source]
+ assert s_w_source.destination is inp_port
- for s in signals:
- inp_port.connect(s)
+ inp_port.disconnect_signal()
- assert inp_port.signal_count() == 1
- assert inp_port.signals[0] == signals[-1]
+ assert inp_port.connected_ports == []
+ assert s_w_source.source.connected_ports == []
+ assert inp_port.signals == []
+ assert s_w_source.source.signals == [s_w_source]
+ assert s_w_source.destination is None
diff --git a/test/port/test_outputport.py b/test/port/test_outputport.py
index 6ca126d5..f48afbdb 100644
--- a/test/port/test_outputport.py
+++ b/test/port/test_outputport.py
@@ -1,18 +1,31 @@
"""
-B-ASIC test suite for InputPort
+B-ASIC test suite for OutputPort
TODO: More info
"""
-from b_asic import OutputPort
+from b_asic import InputPort, OutputPort
import pytest
-def test_connect_multiple_signals(signals):
- """
- test if multiple signals can connect to an output port
- """
- outp_port = OutputPort(0, None)
+@pytest.fixture
+def inp_ports():
+ return [InputPort(_, None) for _ in range(0,3)]
- for s in signals:
- outp_port.connect(s)
+def test_connect_multiple_signals(inp_ports):
+ """Can multiple ports connect to an output port?"""
+ out_port = OutputPort(0, None)
- assert outp_port.signal_count() == 3
- assert outp_port.signals == signals
+ for port in inp_ports:
+ out_port.connect_port(port)
+
+ assert out_port.signal_count() == len(inp_ports)
+
+def test_disconnect_multiple_signals(inp_ports):
+ """Can multiple ports disconnect from an output port?"""
+ out_port = OutputPort(0, None)
+
+ for port in inp_ports:
+ out_port.connect_port(port)
+
+ for _ in inp_ports:
+ out_port.disconnect_signal(0)
+
+ assert out_port.signal_count() == 0
\ No newline at end of file
diff --git a/test/signal/test_signal.py b/test/signal/test_signal.py
new file mode 100644
index 00000000..8c10d1e3
--- /dev/null
+++ b/test/signal/test_signal.py
@@ -0,0 +1,62 @@
+"""
+B-ASIC test suit for the signal module which consists of the Signal class.
+"""
+
+from b_asic.port import InputPort, OutputPort
+from b_asic.signal import Signal
+
+import pytest
+
+def test_signal_creation_and_disconnction_and_connection_changing():
+ in_port = InputPort(0, None)
+ out_port = OutputPort(1, None)
+ s = Signal(out_port, in_port)
+
+ assert in_port.signals == [s]
+ assert out_port.signals == [s]
+ assert s.source is out_port
+ assert s.destination is in_port
+
+ in_port1 = InputPort(0, None)
+ s.connect_destination(in_port1)
+
+ assert in_port.signals == []
+ assert in_port1.signals == [s]
+ assert out_port.signals == [s]
+ assert s.source is out_port
+ assert s.destination is in_port1
+
+ s.disconnect_source()
+
+ assert out_port.signals == []
+ assert in_port1.signals == [s]
+ assert s.source is None
+ assert s.destination is in_port1
+
+ s.disconnect_destination()
+
+ assert out_port.signals == []
+ assert in_port1.signals == []
+ assert s.source is None
+ assert s.destination is None
+
+ out_port1 = OutputPort(0, None)
+ s.connect_source(out_port1)
+
+ assert out_port1.signals == [s]
+ assert s.source is out_port1
+ assert s.destination is None
+
+ s.connect_source(out_port)
+
+ assert out_port.signals == [s]
+ assert out_port1.signals == []
+ assert s.source is out_port
+ assert s.destination is None
+
+ s.connect_destination(in_port)
+
+ assert out_port.signals == [s]
+ assert in_port.signals == [s]
+ assert s.source is out_port
+ assert s.destination is in_port
diff --git a/test/signal_flow_graph/conftest.py b/test/signal_flow_graph/conftest.py
deleted file mode 100644
index 5871ed8e..00000000
--- a/test/signal_flow_graph/conftest.py
+++ /dev/null
@@ -1 +0,0 @@
-import pytest
diff --git a/test/traverse/test_traverse_tree.py b/test/traverse/test_traverse_tree.py
index 2c1d08fe..9f509287 100644
--- a/test/traverse/test_traverse_tree.py
+++ b/test/traverse/test_traverse_tree.py
@@ -24,6 +24,7 @@ def test_traverse_type(large_operation_tree):
def test_traverse_loop(operation_tree):
add_oper_signal = Signal()
- operation_tree._output_ports[0].connect(add_oper_signal)
- operation_tree._input_ports[0].connect(add_oper_signal)
+ operation_tree._output_ports[0].connect_signal(add_oper_signal)
+ operation_tree._input_ports[0].disconnect_signal()
+ operation_tree._input_ports[0].connect_signal(add_oper_signal)
assert len(list(operation_tree.traverse())) == 2
--
GitLab