Merge branch 'integration_branch' into 'develop'

Integrated ID system, traversing and som signal tests

See merge request PUM_TDDD96/B-ASIC!2

b_asic/basic_operation.py

@@ -5,7 +5,7 @@ TODO: More info.
 
 from b_asic.port import InputPort, OutputPort
 from b_asic.signal import SignalSource, SignalDestination
-from b_asic.operation import OperationId, Operation
+from b_asic.operation import Operation
 from b_asic.simulation import SimulationState, OperationState
 from abc import ABC, abstractmethod
 from typing import List, Dict, Optional, Any
@@ -18,16 +18,14 @@ class BasicOperation(Operation):
 	TODO: More info.
 	"""
 
-	_identifier: OperationId
 	_input_ports: List[InputPort]
 	_output_ports: List[OutputPort]
 	_parameters: Dict[str, Optional[Any]]
 
-	def __init__(self, identifier: OperationId):
+	def __init__(self):
 		"""
 		Construct a BasicOperation.
 		"""
-		self._identifier = identifier
 		self._input_ports = []
 		self._output_ports = []
 		self._parameters = {}
@@ -39,9 +37,6 @@ class BasicOperation(Operation):
 		"""
 		pass
 
-	def identifier(self) -> OperationId:
-		return self._identifier
-
 	def inputs(self) -> List[InputPort]:
 		return self._input_ports.copy()
 
@@ -102,4 +97,13 @@ class BasicOperation(Operation):
 			return results
 		return [self]
 
+	@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]
+
+		return neighbours
+
 	# TODO: More stuff.

b_asic/core_operations.py

@@ -4,8 +4,9 @@ TODO: More info.
 """
 
 from b_asic.port import InputPort, OutputPort
-from b_asic.operation import OperationId, Operation
+from b_asic.operation import Operation
 from b_asic.basic_operation import BasicOperation
+from b_asic.graph_id import GraphIDType
 from numbers import Number
 
 
@@ -15,7 +16,7 @@ class Input(Operation):
 	TODO: More info.
 	"""
 
-	# TODO: Implement.
+	# TODO: Implement all functions.
 	pass
 
 
@@ -25,17 +26,19 @@ class Constant(BasicOperation):
 	TODO: More info.
 	"""
 
-	def __init__(self, identifier: OperationId, value: Number):
+	def __init__(self, value: Number):
 		"""
 		Construct a Constant.
 		"""
-		super().__init__(identifier)
-		self._output_ports = [OutputPort()] # TODO: Generate appropriate ID for ports.
+		super().__init__()
+		self._output_ports = [OutputPort(1)] # TODO: Generate appropriate ID for ports.
 		self._parameters["value"] = value
 
 	def evaluate(self, inputs: list) -> list:
 		return [self.param("value")]
 
+	def get_op_name(self) -> GraphIDType:
+		return "const"
 
 class Addition(BasicOperation):
 	"""
@@ -43,17 +46,20 @@ class Addition(BasicOperation):
 	TODO: More info.
 	"""
 
-	def __init__(self, identifier: OperationId):
+	def __init__(self):
 		"""
 		Construct an Addition.
 		"""
-		super().__init__(identifier)
-		self._input_ports = [InputPort(), InputPort()] # TODO: Generate appropriate ID for ports.
-		self._output_ports = [OutputPort()] # TODO: Generate appropriate ID for ports.
+		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.
 
 	def evaluate(self, inputs: list) -> list:
 		return [inputs[0] + inputs[1]]
 
+	def get_op_name(self) -> GraphIDType:
+		return "add"
+
 
 class ConstantMultiplication(BasicOperation):
 	"""
@@ -61,16 +67,19 @@ class ConstantMultiplication(BasicOperation):
 	TODO: More info.
 	"""
 
-	def __init__(self, identifier: OperationId, coefficient: Number):
+	def __init__(self, coefficient: Number):
 		"""
 		Construct a ConstantMultiplication.
 		"""
-		super().__init__(identifier)
-		self._input_ports = [InputPort()] # TODO: Generate appropriate ID for ports.
-		self._output_ports = [OutputPort()] # TODO: Generate appropriate ID for ports.
+		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._parameters["coefficient"] = coefficient
 
 	def evaluate(self, inputs: list) -> list:
 		return [inputs[0] * self.param("coefficient")]
 
+	def get_op_name(self) -> GraphIDType:
+		return "const_mul"
+
 # TODO: More operations.

b_asic/graph_id.py

+"""
+B-ASIC Graph ID module for handling IDs of different objects in a graph.
+TODO: More info
+"""
+
+from collections import defaultdict
+from typing import NewType, Union, DefaultDict
+
+GraphID = NewType("GraphID", str)
+GraphIDType = NewType("GraphIDType", str)
+GraphIDNumber = NewType("GraphIDNumber", int)
+
+class GraphIDGenerator:
+    """
+    A class that generates Graph IDs for objects.
+    """
+
+    _next_id_number: DefaultDict[GraphIDType, GraphIDNumber]
+
+    def __init__(self):
+        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.
+        """
+        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
+

b_asic/operation.py

@@ -10,9 +10,7 @@ from typing import NewType, List, Dict, Optional, Any, TYPE_CHECKING
 if TYPE_CHECKING:
 	from b_asic.port import InputPort, OutputPort
 	from b_asic.simulation import SimulationState
-
-OperationId = NewType("OperationId", int)
-
+	from b_asic.graph_id import GraphIDType
 
 class Operation(ABC):
 	"""
@@ -20,14 +18,6 @@ class Operation(ABC):
 	TODO: More info.
 	"""
 
-	@abstractmethod
-	def identifier(self) -> OperationId:
-		"""
-		Get the unique identifier.
-		TODO: Move id info to SFG, remove id class members.
-		"""
-		pass
-
 	@abstractmethod
 	def inputs(self) -> "List[InputPort]":
 		"""
@@ -109,5 +99,17 @@ class Operation(ABC):
 		"""
 		pass
 
+	@abstractmethod
+	def get_op_name(self) -> "GraphIDType":
+		"""Returns a string representing the operation name of the operation."""
+		pass
+
+	@abstractmethod
+	def neighbours(self) -> "List[Operation]":
+		"""
+		Return all operations that are connected by signals to this operation.
+		If no neighbours are found this returns an empty list
+		"""
+
 	# TODO: More stuff.
 

b_asic/signal.py

@@ -6,9 +6,6 @@ TODO: More info.
 from b_asic.operation import Operation
 from typing import NewType
 
-SignalId = NewType("SignalId", int)
-
-
 class SignalSource:
 	"""
 	Handle to a signal source.
@@ -50,22 +47,14 @@ class Signal:
 	A connection between two operations consisting of a source and destination handle.
 	TODO: More info.
 	"""
-	_identifier: SignalId
 	source: SignalSource
 	destination: SignalDestination
 
-	def __init__(self, identifier: SignalId, source: SignalSource, destination: SignalDestination):
+	def __init__(self, source: SignalSource, destination: SignalDestination):
 		"""
 		Construct a Signal.
 		"""
-		self._identifier = identifier
 		self.source = source
 		self.destination = destination
 
-	def identifier(self) -> SignalId:
-		"""
-		Get the unique identifier.
-		"""
-		return self._identifier
-
 	# TODO: More stuff.

b_asic/signal_flow_graph.py

@@ -3,12 +3,13 @@ B-ASIC Signal Flow Graph Module.
 TODO: More info.
 """
 
-from b_asic.operation import OperationId, Operation
+from b_asic.operation import Operation
 from b_asic.basic_operation import BasicOperation
-from b_asic.signal import SignalSource, SignalDestination
+from b_asic.signal import Signal, SignalSource, SignalDestination
 from b_asic.simulation import SimulationState, OperationState
-from typing import List
+from b_asic.graph_id import GraphIDGenerator, GraphID
 
+from typing import List, Dict, Union, Optional
 
 class SFG(BasicOperation):
 	"""
@@ -16,22 +17,60 @@ class SFG(BasicOperation):
 	TODO: More info.
 	"""
 
-	_operations: List[Operation]
+	_graph_objects_by_id: Dict[GraphID, Union[Operation, Signal]]
+	_graph_id_generator: GraphIDGenerator
 
-	def __init__(self, identifier: OperationId, input_destinations: List[SignalDestination], output_sources: List[SignalSource]):
-		"""
-		Construct a SFG.
-		"""
-		super().__init__(identifier)
+	def __init__(self, input_destinations: List[SignalDestination], output_sources: List[SignalSource]):
+		"""Constructs an SFG."""
+		super().__init__()
 		# TODO: Allocate input/output ports with appropriate IDs.
-		self._operations = []
+		
+		self._graph_objects_by_id = dict # Map Operation ID to Operation objects
+		self._graph_id_generator = GraphIDGenerator() 
+
 		# TODO: Traverse the graph between the inputs/outputs and add to self._operations.
 		# TODO: Connect ports with signals with appropriate IDs.
 
 	def evaluate(self, inputs: list) -> list:
 		return [] # TODO: Implement
 
-	def split(self) -> List[Operation]:
-		return self._operations
+	def add_operation(self, operation: Operation) -> GraphID:
+		"""Adds the entered operation to the SFG's dictionary of graph objects and
+	 	returns a generated GraphID for it.
+		
+		Keyword arguments:
+		operation: Operation to add to the graph.
+		"""
+		return self._add_graph_obj(operation, operation.get_op_name())
+
+
+	def add_signal(self, signal: Signal) -> GraphID:
+		"""Adds the entered signal to the SFG's dictionary of graph objects and returns
+		a generated GraphID for it.
+		
+		Keyword argumentst:
+		signal: Signal to add to the graph.
+		"""
+		return self._add_graph_obj(signal, 'sig')
+
+
+	def find_by_id(self, graph_id: GraphID) -> Optional[Operation]:
+		"""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.
+
+		Keyword arguments:
+		graph_id: Graph ID of the wanted object.
+		"""
+		if graph_id in self._graph_objects_by_id:
+			return self._graph_objects_by_id[graph_id]
+		else:
+			return None
+			
+
+
+	def _add_graph_obj(self, obj: Union[Operation, Signal], operation_id_type: str):
+		graph_id = self._graph_id_generator.get_next_id(operation_id_type)
+		self._graph_objects_by_id[graph_id] = obj
+		return graph_id 
+
 
-	# TODO: More stuff.

b_asic/simulation.py

@@ -3,7 +3,6 @@ B-ASIC Simulation Module.
 TODO: More info.
 """
 
-from b_asic.operation import OperationId
 from numbers import Number
 from typing import List, Dict
 
@@ -31,7 +30,7 @@ class SimulationState:
 	TODO: More info.
 	"""
 
-	operation_states: Dict[OperationId, OperationState]
+	# operation_states: Dict[OperationId, OperationState]
 	iteration: int
 
 	def __init__(self):

b_asic/traverse_tree.py

+"""
+B-ASIC Operation Tree Traversing Module.
+TODO:
+    - Get a first operation or? an entire operation tree
+    - For each start point, follow it to the next operation from it's out port.
+    - If we are searching for a specific operation end.
+    - If we are searching for a specific type of operation add the operation to a list and continue.
+    - When we no more out ports can be traversed return results and end.
+"""
+
+from typing import List, Optional
+from collections import deque
+
+from b_asic.operation import Operation
+
+
+class Traverse:
+    """Traverse operation tree.
+    TODO:
+        - More info.
+        - Check if a datastructure other than list suits better as return value.
+        - Implement the type check for operation.
+    """
+
+    def __init__(self, operation: Operation):
+        """Construct a TraverseTree."""
+        self._initial_operation = operation
+
+    def _breadth_first_search(self, start: Operation) -> List[Operation]:
+        """Use breadth first search to traverse the operation tree."""
+        visited: List[Operation] = [start]
+        queue = deque([start])
+        while queue:
+            operation = queue.popleft()
+            for n_operation in operation.neighbours:
+                if n_operation not in visited:
+                    visited.append(n_operation)
+                    queue.append(n_operation)
+
+        return visited
+
+    def traverse(self, type_: Optional[Operation] = None) -> List[Operation]:
+        """Traverse the the operation tree and return operation where type matches.
+        If the type is None then return the entire tree.
+
+        Keyword arguments:
+        type_-- the operation type to search for (default None)
+        """
+
+        operations: List[Operation] = self._breadth_first_search(self._initial_operation)
+        if type_ is not None:
+            operations = [oper for oper in operations if isinstance(oper, type_)]
+
+        return operations

test/fixtures/signal.py

@@ -6,15 +6,15 @@ Use a fixture for initializing objects and pass them as argument to a test funct
 """
 @pytest.fixture
 def signal():
-    source = SignalSource(Addition(0), 1)
-    dest = SignalDestination(Addition(1), 2)
-    return Signal(0, source, dest)
+    source = SignalSource(Addition(), 1)
+    dest = SignalDestination(Addition(), 2)
+    return Signal(source, dest)
 
 @pytest.fixture
 def signals():
     ret = []
-    for i in range(0,3):
-        source = SignalSource(Addition(0), 1)
-        dest = SignalDestination(Addition(1), 2)
-        ret.append(Signal(i, source, dest))
+    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

test/graph_id/conftest.py

+import pytest

test/graph_id/test_graph_id_generator.py

+"""
+B-ASIC test suite for graph id generator.
+"""
+
+from b_asic.graph_id import GraphIDGenerator, GraphID
+
+import pytest
+
+def test_empty_string_generator():
+    """Test the graph id generator for an empty string type."""
+    graph_id_generator = GraphIDGenerator()
+    assert graph_id_generator.get_next_id("") == "1"
+    assert graph_id_generator.get_next_id("") == "2"
+
+
+def test_normal_string_generator():
+    """"Test the graph id generator for a normal string type."""
+    graph_id_generator = GraphIDGenerator()
+    assert graph_id_generator.get_next_id("add") == "add1"
+    assert graph_id_generator.get_next_id("add") == "add2"
+
+def test_different_strings_generator():
+    """Test the graph id generator for different strings."""
+    graph_id_generator = GraphIDGenerator()
+    assert graph_id_generator.get_next_id("sub") == "sub1"
+    assert graph_id_generator.get_next_id("mul") == "mul1"
+    assert graph_id_generator.get_next_id("sub") == "sub2"
+    assert graph_id_generator.get_next_id("mul") == "mul2"
+    
\ No newline at end of file

test/port/test_port.py

@@ -13,10 +13,10 @@ def test_connect_one_signal_to_port(signal):
     assert port.signal() == signal
 
 def test_change_port_signal():
-    source = SignalSource(Addition(0), 1)
-    dest = SignalDestination(Addition(1),2)
-    signal1 = Signal(1, source, dest)
-    signal2 = Signal(2, source, dest)
+    source = SignalSource(Addition, 1)
+    dest = SignalDestination(Addition,2)
+    signal1 = Signal(source, dest)
+    signal2 = Signal(source, dest)
 
     port = InputPort(0)
     port.connect(signal1)

test/signal_flow_graph/conftest.py

+import pytest

test/signal_flow_graph/test_signal_flow_graph.py

+from b_asic.signal_flow_graph import SFG
+from b_asic.core_operations import Addition, Constant
+from b_asic.signal import Signal

test/traverse/test_traverse_tree.py

+"""
+TODO:
+    - Rewrite to more clean code, not so repetitive
+    - Update when signals and id's has been merged.
+"""
+
+from b_asic.core_operations import Constant, Addition
+from b_asic.signal import Signal, SignalSource, SignalDestination
+from b_asic.port import InputPort, OutputPort
+from b_asic.traverse_tree import Traverse
+
+import pytest
+
+@pytest.fixture
+def operation():
+    return Constant(2)
+
+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._output_ports[0].connect(oper_signal)
+    dest_oper._input_ports[index].connect(oper_signal)
+    return oper
+
+@pytest.fixture
+def operation_tree():
+    add_oper = Addition()
+
+    const_oper = create_operation(Constant, add_oper, 0, value=2)
+    const_oper_2 = create_operation(Constant, add_oper, 1, value=3)
+
+    return add_oper
+
+@pytest.fixture
+def large_operation_tree():
+    add_oper = Addition()
+    add_oper_2 = Addition()
+
+    const_oper = create_operation(Constant, add_oper, 0, value=2)
+    const_oper_2 = create_operation(Constant, add_oper, 1, value=3)
+
+    const_oper_3 = create_operation(Constant, add_oper_2, 0, value=4)
+    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._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._output_ports[0].connect(add_oper_2_signal)
+    add_oper_3._input_ports[1].connect(add_oper_2_signal)
+    return const_oper
+
+def test_traverse_single_tree(operation):
+    traverse = Traverse(operation)
+    assert traverse.traverse() == [operation]
+
+def test_traverse_tree(operation_tree):
+    traverse = Traverse(operation_tree)
+    assert len(traverse.traverse()) == 3
+
+def test_traverse_large_tree(large_operation_tree):
+    traverse = Traverse(large_operation_tree)
+    assert len(traverse.traverse()) == 7
+
+def test_traverse_type(large_operation_tree):
+    traverse = Traverse(large_operation_tree)
+    assert len(traverse.traverse(Addition)) == 3
+    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)
+    operation_tree._output_ports[0].connect(add_oper_signal)
+    operation_tree._input_ports[0].connect(add_oper_signal)
+    traverse = Traverse(operation_tree)
+    assert len(traverse.traverse()) == 2
\ No newline at end of file