Merge branch '45-signal-interface' into 'develop'

Resolve "Signal Interface"

See merge request PUM_TDDD96/B-ASIC!5

b_asic/basic_operation.py

@@ -8,7 +8,7 @@ from typing import List, Dict, Optional, Any
 from numbers import Number
 
 from b_asic.port import InputPort, OutputPort
-from b_asic.signal import SignalSource, SignalDestination
+from b_asic.signal import Signal
 from b_asic.operation import Operation
 from b_asic.simulation import SimulationState, OperationState
 
@@ -73,7 +73,7 @@ class BasicOperation(Operation):
 		while self_state.iteration < state.iteration:
 			input_values: List[Number] = [0] * input_count
 			for i in range(input_count):
-				source: SignalSource = self._input_ports[i].signal().source
+				source: Signal = self._input_ports[i].signal
 				input_values[i] = source.operation.evaluate_outputs(state)[source.port_index]
 
 			self_state.output_values = self.evaluate(input_values)
@@ -81,7 +81,7 @@ class BasicOperation(Operation):
 			self_state.iteration += 1
 			for i in range(output_count):
 				for signal in self._output_ports[i].signals():
-					destination: SignalDestination = signal.destination
+					destination: Signal = signal.destination
 					destination.evaluate_outputs(state)
 
 		return self_state.output_values
@@ -96,9 +96,13 @@ class BasicOperation(Operation):
 	@property
 	def neighbours(self) -> List[Operation]:
 		neighbours: List[Operation] = []
-		for port in self._output_ports + self._input_ports:
-			for signal in port.signals():
-				neighbours += [signal.source.operation, signal.destination.operation]
+		for port in self._input_ports:
+			for signal in port.signals:
+				neighbours.append(signal.source.operation)
+
+		for port in self._output_ports:
+			for signal in port.signals:
+				neighbours.append(signal.destination.operation)
 
 		return neighbours
 

b_asic/core_operations.py

@@ -28,7 +28,7 @@ class Constant(BasicOperation):
 	def __init__(self, value: Number):
 		"""Construct a Constant."""
 		super().__init__()
-		self._output_ports = [OutputPort(1)] # TODO: Generate appropriate ID for ports.
+		self._output_ports = [OutputPort(1, self)] # TODO: Generate appropriate ID for ports.
 		self._parameters["value"] = value
 
 	def evaluate(self, inputs: list) -> list:
@@ -45,8 +45,8 @@ class Addition(BasicOperation):
 	def __init__(self):
 		"""Construct an Addition."""
 		super().__init__()
-		self._input_ports = [InputPort(1), InputPort(1)] # TODO: Generate appropriate ID for ports.
-		self._output_ports = [OutputPort(1)] # TODO: Generate appropriate ID for ports.
+		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 evaluate(self, inputs: list) -> list:
 		return [inputs[0] + inputs[1]]
@@ -63,8 +63,8 @@ class ConstantMultiplication(BasicOperation):
 	def __init__(self, coefficient: Number):
 		"""Construct a ConstantMultiplication."""
 		super().__init__()
-		self._input_ports = [InputPort(1)] # TODO: Generate appropriate ID for ports.
-		self._output_ports = [OutputPort(1)] # TODO: Generate appropriate ID for ports.
+		self._input_ports = [InputPort(1), self] # TODO: Generate appropriate ID for ports.
+		self._output_ports = [OutputPort(1, self)] # TODO: Generate appropriate ID for ports.
 		self._parameters["coefficient"] = coefficient
 
 	def evaluate(self, inputs: list) -> list:

b_asic/port.py

@@ -7,6 +7,7 @@ from abc import ABC, abstractmethod
 from typing import NewType, Optional, List
 
 from b_asic.signal import Signal
+from b_asic.operation import Operation
 
 PortId = NewType("PortId", int)
 
@@ -16,29 +17,38 @@ class Port(ABC):
 	TODO: More info.
 	"""
 
-	_identifier: PortId
+	_port_id: PortId
+	_operation: Operation
 
-	def __init__(self, identifier: PortId):
-		"""Construct a Port."""
-		self._identifier = identifier
+	def __init__(self, port_id: PortId, operation: Operation):
+		self._port_id = port_id
+		self._operation = operation
 
+	@property
 	def identifier(self) -> PortId:
 		"""Get the unique identifier."""
-		return self._identifier
+		return self._port_id
 
+	@property
+	def operation(self) -> Operation:
+		"""Get the connected operation."""
+		return self._operation
+
+	@property
 	@abstractmethod
 	def signals(self) -> List[Signal]:
 		"""Get a list of all connected signals."""
 		pass
 
+	@property
 	@abstractmethod
-	def signal_count(self) -> int:
-		"""Get the number of connected signals."""
+	def signal(self, i: int = 0) -> Signal:
+		"""Get the connected signal at index i."""
 		pass
 
 	@abstractmethod
-	def signal(self, i: int = 0) -> Signal:
-		"""Get the connected signal at index i."""
+	def signal_count(self) -> int:
+		"""Get the number of connected signals."""
 		pass
 
 	@abstractmethod
@@ -51,6 +61,7 @@ class Port(ABC):
 		"""Disconnect a signal."""
 		pass
 
+
 	# TODO: More stuff.
 
 
@@ -60,26 +71,30 @@ class InputPort(Port):
 	"""
 	_source_signal: Optional[Signal]
 
-	def __init__(self, identifier: PortId):
-		super().__init__(identifier)
+	def __init__(self, port_id: PortId, operation: Operation):
+		super().__init__(port_id, operation)
 		self._source_signal = None
 
+	@property
 	def signals(self) -> List[Signal]:
 		return [] if self._source_signal is None else [self._source_signal]
 
-	def signal_count(self) -> int:
-		return 0 if self._source_signal is None else 1
-
+	@property
 	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
 
+	def signal_count(self) -> int:
+		return 0 if self._source_signal is None else 1
+
 	def connect(self, signal: Signal) -> None:
 		self._source_signal = signal
+		signal.destination = self
 
 	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
 
 	# TODO: More stuff.
@@ -92,23 +107,26 @@ class OutputPort(Port):
 
 	_destination_signals: List[Signal]
 
-	def __init__(self, identifier: PortId):
-		super().__init__(identifier)
+	def __init__(self, port_id: PortId, operation: Operation):
+		super().__init__(port_id, operation)
 		self._destination_signals = []
 
+	@property
 	def signals(self) -> List[Signal]:
 		return self._destination_signals.copy()
 
-	def signal_count(self) -> int:
-		return len(self._destination_signals)
-
+	@property
 	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)
+
 	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.

b_asic/signal.py

 """@package docstring
 B-ASIC Signal Module.
-TODO: More info.
 """
+from typing import TYPE_CHECKING, Optional
+if TYPE_CHECKING:
+	from b_asic import OutputPort, InputPort
 
-from b_asic.operation import Operation
-
-
-class SignalSource:
-	"""Handle to a signal source.
-	TODO: More info.
-	"""
-	operation: Operation
-	port_index: int
-
-	def __init__(self, operation: Operation, port_index: int):
-		self.operation = operation
-		self.port_index = port_index
-
-	# TODO: More stuff.
+class Signal:
+	"""A connection between two ports."""
+	_source: "OutputPort"
+	_destination: "InputPort"
 
+	def __init__(self, src: Optional["OutputPort"] = None, dest: Optional["InputPort"] = None):
+		self._source = src
+		self._destination = dest
 
-class SignalDestination:
-	"""Handle to a signal destination.
-	TODO: More info.
-	"""
-	operation: Operation
-	port_index: int
+	@property	
+	def source(self) -> "InputPort":
+		return self._source
 
-	def __init__(self, operation: Operation, port_index: int):
-		self.operation = operation
-		self.port_index = port_index
+	@property
+	def destination(self) -> "OutputPort":
+		return self._destination
 
-	# TODO: More stuff.
+	@source.setter
+	def source(self, src: "Outputport") -> None:
+		self._source = src
 
+	@destination.setter
+	def destination(self, dest: "InputPort") -> None:
+		self._destination = dest
 
-class Signal:
-	"""A connection between two operations consisting of a source and destination handle.
-	TODO: More info.
-	"""
-	source: SignalSource
-	destination: SignalDestination
+	def disconnect_source(self) -> None:
+		self._source = None
 
-	def __init__(self, source: SignalSource, destination: SignalDestination):
-		self.source = source
-		self.destination = destination
+	def disconnect_destination(self) -> None:
+		self._destination = None
 
-	# TODO: More stuff.

b_asic/signal_flow_graph.py

@@ -7,7 +7,9 @@ from typing import List, Dict, Union, Optional
 
 from b_asic.operation import Operation
 from b_asic.basic_operation import BasicOperation
-from b_asic.signal import Signal, SignalSource, SignalDestination
+from b_asic.signal import Signal
+from b_asic.simulation import SimulationState, OperationState
+from typing import List
 from b_asic.graph_id import GraphIDGenerator, GraphID
 
 
@@ -19,7 +21,7 @@ class SFG(BasicOperation):
 	_graph_objects_by_id: Dict[GraphID, Union[Operation, Signal]]
 	_graph_id_generator: GraphIDGenerator
 
-	def __init__(self, input_destinations: List[SignalDestination], output_sources: List[SignalSource]):
+	def __init__(self, input_destinations: List[Signal], output_sources: List[Signal]):
 		super().__init__()
 		# TODO: Allocate input/output ports with appropriate IDs.
 

test/fixtures/signal.py

 import pytest
-from b_asic import Signal, SignalSource, SignalDestination, Addition
+from b_asic import Signal
 
-"""
-Use a fixture for initializing objects and pass them as argument to a test function
-"""
 @pytest.fixture
 def signal():
-    source = SignalSource(Addition(), 1)
-    dest = SignalDestination(Addition(), 2)
-    return Signal(source, dest)
+    return Signal()
 
 @pytest.fixture
 def signals():
-    ret = []
-    for _ in range(0,3):
-        source = SignalSource(Addition(), 1)
-        dest = SignalDestination(Addition(), 2)
-        ret.append(Signal(source, dest))
-    return ret
\ No newline at end of file
+    return [Signal() for _ in range(0,3)]

test/port/test_inputport.py

@@ -2,22 +2,19 @@
 B-ASIC test suite for Inputport
 """
 
-# import module we are testing
 from b_asic import InputPort
 
-# import dependencies
-from b_asic import Signal, SignalSource, SignalDestination, Addition
-
 import pytest
 
 def test_connect_multiple_signals(signals):
     """
     test if only one signal can connect to an input port
     """
-    inp_port = InputPort(0)
+    inp_port = InputPort(0, None)
 
     for s in signals:
         inp_port.connect(s)
 
     assert inp_port.signal_count() == 1
-    assert inp_port.signals()[0] == signals[-1]
\ No newline at end of file
+    assert inp_port.signals[0] == signals[-1]
+

test/port/test_outputport.py

@@ -2,17 +2,17 @@
 B-ASIC test suite for InputPort
 TODO: More info
 """
-from b_asic import OutputPort, Signal, SignalSource, SignalDestination, Addition
+from b_asic import OutputPort
 import pytest
 
 def test_connect_multiple_signals(signals):
     """
     test if multiple signals can connect to an output port
     """
-    outp_port = OutputPort(0)
+    outp_port = OutputPort(0, None)
 
     for s in signals:
         outp_port.connect(s)
 
     assert outp_port.signal_count() == 3
-    assert outp_port.signals() == signals
\ No newline at end of file
+    assert outp_port.signals == signals
\ No newline at end of file

test/port/test_port.py

-"""
-B-ASIC test suite for Port interface, place all general test cases for abstract class Port here
-"""
-
-from b_asic import InputPort, OutputPort, Signal, SignalSource, SignalDestination, Addition
-import pytest
-
-
-def test_connect_one_signal_to_port(signal):
-    port = InputPort(0)
-    port.connect(signal)
-    assert len(port.signals()) == 1
-    assert port.signal() == signal
-
-def test_change_port_signal():
-    source = SignalSource(Addition, 1)
-    dest = SignalDestination(Addition,2)
-    signal1 = Signal(source, dest)
-    signal2 = Signal(source, dest)
-
-    port = InputPort(0)
-    port.connect(signal1)
-    assert port.signal() == signal1
-    port.connect(signal2)
-    assert port.signal() == signal2
\ No newline at end of file

test/traverse/test_traverse_tree.py

@@ -5,7 +5,7 @@ TODO:
 """
 
 from b_asic.core_operations import Constant, Addition
-from b_asic.signal import Signal, SignalSource, SignalDestination
+from b_asic.signal import Signal 
 from b_asic.port import InputPort, OutputPort
 from b_asic.traverse_tree import Traverse
 
@@ -17,10 +17,9 @@ def operation():
 
 def create_operation(_type, dest_oper, index, **kwargs):
     oper = _type(**kwargs)
-    oper_signal_source = SignalSource(oper, 0)
-    oper_signal_dest = SignalDestination(dest_oper, index)
-    oper_signal = Signal(oper_signal_source, oper_signal_dest)
+    oper_signal = Signal()
     oper._output_ports[0].connect(oper_signal)
+
     dest_oper._input_ports[index].connect(oper_signal)
     return oper
 
@@ -45,15 +44,11 @@ def large_operation_tree():
     const_oper_4 = create_operation(Constant, add_oper_2, 1, value=5)
 
     add_oper_3 = Addition()
-    add_oper_signal_source = SignalSource(add_oper, 0)
-    add_oper_signal_dest = SignalDestination(add_oper_3, 0)
-    add_oper_signal = Signal(add_oper_signal_source, add_oper_signal_dest)
+    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_2_signal_source = SignalSource(add_oper_2, 0)
-    add_oper_2_signal_dest = SignalDestination(add_oper_3, 1)
-    add_oper_2_signal = Signal(add_oper_2_signal_source, add_oper_2_signal_dest)
+    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)
     return const_oper
@@ -76,9 +71,7 @@ def test_traverse_type(large_operation_tree):
     assert len(traverse.traverse(Constant)) == 4
 
 def test_traverse_loop(operation_tree):
-    add_oper_signal_source = SignalSource(operation_tree, 0)
-    add_oper_signal_dest = SignalDestination(operation_tree, 0)
-    add_oper_signal = Signal(add_oper_signal_source, add_oper_signal_dest)
+    add_oper_signal = Signal()
     operation_tree._output_ports[0].connect(add_oper_signal)
     operation_tree._input_ports[0].connect(add_oper_signal)
     traverse = Traverse(operation_tree)