diff --git a/b_asic/basic_operation.py b/b_asic/basic_operation.py index 547b3bf32006caf1e88508163528ada285599958..b9b0895eca5c3db96a75f2c8a5b344fe3bf557df 100644 --- a/b_asic/basic_operation.py +++ b/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 from b_asic.utilities import breadth_first_search @@ -74,7 +74,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) @@ -82,7 +82,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 @@ -97,9 +97,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 diff --git a/b_asic/core_operations.py b/b_asic/core_operations.py index 513fe7cfb7bacc73afa3e061651e17dd20d415ec..45919b8bdb4f892cf9ff30a8305d585678b22220 100644 --- a/b_asic/core_operations.py +++ b/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: diff --git a/b_asic/port.py b/b_asic/port.py index f67defb79166432b79333abfdd161c2e09c1a8ba..4c6fb244b5882e4a6bbc9cadc9e4d38016bc5748 100644 --- a/b_asic/port.py +++ b/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. diff --git a/b_asic/signal.py b/b_asic/signal.py index 7e63ebfb331081b268b6e7d8ea7e9a1d375c29df..17078138e5ff75889cd6afe8759584f04749edfa 100644 --- a/b_asic/signal.py +++ b/b_asic/signal.py @@ -1,48 +1,38 @@ """@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. diff --git a/b_asic/signal_flow_graph.py b/b_asic/signal_flow_graph.py index 914cf390449cac7e7262ef485c8da36ccc045a53..f7d4be640dee3605855f02a159c54b574774e565 100644 --- a/b_asic/signal_flow_graph.py +++ b/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. diff --git a/test/fixtures/__init__.py b/test/fixtures/__init__.py deleted file mode 100644 index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..0000000000000000000000000000000000000000 diff --git a/test/fixtures/operation_tree.py b/test/fixtures/operation_tree.py index d02679b54da6a51fb066a15c3738f192a01fc38b..bcc0bb27fac9fa8d1a024d67373469c52f8a45dc 100644 --- a/test/fixtures/operation_tree.py +++ b/test/fixtures/operation_tree.py @@ -1,20 +1,26 @@ from b_asic.core_operations import Addition, Constant -from b_asic.signal import Signal, SignalDestination, SignalSource +from b_asic.signal import Signal 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_signal = Signal() oper._output_ports[0].connect(oper_signal) + dest_oper._input_ports[index].connect(oper_signal) return oper @pytest.fixture -def operation(): - return Constant(2) +def operation_tree(): + add_oper = Addition() + create_operation(Constant, add_oper, 0, value=2) + create_operation(Constant, add_oper, 1, value=3) + return add_oper @pytest.fixture def large_operation_tree(): @@ -28,31 +34,11 @@ def large_operation_tree(): 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 - -@pytest.fixture -def addition_tree(): - add_oper = Addition() - create_operation(Constant, add_oper, 0, value=2) - create_operation(Constant, add_oper, 1, value=3) - return add_oper - -@pytest.fixture -def looping_addition_tree(addition_tree): - add_oper_signal_source = SignalSource(addition_tree, 0) - add_oper_signal_dest = SignalDestination(addition_tree, 0) - add_oper_signal = Signal(add_oper_signal_source, add_oper_signal_dest) - addition_tree._output_ports[0].connect(add_oper_signal) - addition_tree._input_ports[0].connect(add_oper_signal) - return addition_tree \ No newline at end of file diff --git a/test/fixtures/signal.py b/test/fixtures/signal.py index 64b96f5541c7866c4540cd3052a5cbeaea5e4400..9139e93a529cc7a371426b9b97b4d31ddf95d2f7 100644 --- a/test/fixtures/signal.py +++ b/test/fixtures/signal.py @@ -1,20 +1,10 @@ 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)] diff --git a/test/port/__init__.py b/test/port/__init__.py deleted file mode 100644 index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..0000000000000000000000000000000000000000 diff --git a/test/port/test_inputport.py b/test/port/test_inputport.py index d761900a500979d09a7b2584a9249eefda914d27..f0e70cb761ddcd76b2144b1f2c26e606922d213e 100644 --- a/test/port/test_inputport.py +++ b/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] + diff --git a/test/port/test_outputport.py b/test/port/test_outputport.py index 5f7b8f49f856e891fa6e70ce77cc19e464f6954d..5c76bb480fa63488073f6dab9f82c3f3ce00b4f3 100644 --- a/test/port/test_outputport.py +++ b/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 diff --git a/test/port/test_port.py b/test/port/test_port.py deleted file mode 100644 index 7e1fc9b7589a4955966d3333d336bdb6f3245014..0000000000000000000000000000000000000000 --- a/test/port/test_port.py +++ /dev/null @@ -1,25 +0,0 @@ -""" -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 diff --git a/test/traverse/test_traverse_tree.py b/test/traverse/test_traverse_tree.py index 125b68f2c49b52b5152ebef80f2072f0837dce05..2c1d08fe555df06ff86845812ca5df4fef4b5c92 100644 --- a/test/traverse/test_traverse_tree.py +++ b/test/traverse/test_traverse_tree.py @@ -1,4 +1,6 @@ from b_asic.core_operations import Constant, Addition +from b_asic.signal import Signal +from b_asic.port import InputPort, OutputPort import pytest @@ -7,20 +9,21 @@ def test_traverse_single_tree(operation): constant = Constant(None) assert list(constant.traverse()) == [constant] -def test_traverse_tree(addition_tree): +def test_traverse_tree(operation_tree): """Traverse a basic addition tree with two constants.""" - assert len(list(addition_tree.traverse())) == 3 + assert len(list(operation_tree.traverse())) == 3 def test_traverse_large_tree(large_operation_tree): """Traverse a larger tree.""" assert len(list(large_operation_tree.traverse())) == 7 def test_traverse_type(large_operation_tree): - """Traverse the operation tree and return operations of specific type.""" - operations = list(large_operation_tree.traverse()) - assert len(list(filter(lambda type_: isinstance(type_, Addition), operations))) == 3 - assert len(list(filter(lambda type_: isinstance(type_, Constant), operations))) == 4 + traverse = list(large_operation_tree.traverse()) + assert len(list(filter(lambda type_: isinstance(type_, Addition), traverse))) == 3 + assert len(list(filter(lambda type_: isinstance(type_, Constant), traverse))) == 4 -def test_traverse_loop(looping_addition_tree): - """Traverse two operations coupled to eachother in a loop.""" - assert len(list(looping_addition_tree.traverse())) == 2 \ No newline at end of file +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) + assert len(list(operation_tree.traverse())) == 2