From 62b1947577f7cded628fcbb39c23c6af24c85ec2 Mon Sep 17 00:00:00 2001
From: angloth <angus.lothian@hotmail.com>
Date: Wed, 18 Mar 2020 17:16:29 +0100
Subject: [PATCH] Move abstract operation to operation module and move utilties
implementation to operation, remove abstract operation and utilities module.
---
b_asic/__init__.py | 2 -
b_asic/abstract_operation.py | 114 ---------------------------------
b_asic/core_operations.py | 2 +-
b_asic/operation.py | 118 ++++++++++++++++++++++++++++++++++-
b_asic/signal_flow_graph.py | 2 +-
b_asic/utilities.py | 21 -------
6 files changed, 118 insertions(+), 141 deletions(-)
delete mode 100644 b_asic/abstract_operation.py
delete mode 100644 b_asic/utilities.py
diff --git a/b_asic/__init__.py b/b_asic/__init__.py
index 67e1c856..7e40ad52 100644
--- a/b_asic/__init__.py
+++ b/b_asic/__init__.py
@@ -2,7 +2,6 @@
Better ASIC Toolbox.
TODO: More info.
"""
-from b_asic.abstract_operation import *
from b_asic.core_operations import *
from b_asic.graph_component import *
from b_asic.graph_id import *
@@ -13,4 +12,3 @@ from b_asic.schema import *
from b_asic.signal_flow_graph import *
from b_asic.signal import *
from b_asic.simulation import *
-from b_asic.utilities import *
diff --git a/b_asic/abstract_operation.py b/b_asic/abstract_operation.py
deleted file mode 100644
index 68b3c02a..00000000
--- a/b_asic/abstract_operation.py
+++ /dev/null
@@ -1,114 +0,0 @@
-"""@package docstring
-B-ASIC Abstract Operation Module.
-TODO: More info
-"""
-
-from abc import abstractmethod
-from typing import List, Dict, Optional, Any
-from numbers import Number
-
-from b_asic.operation import Operation
-from b_asic.graph_component import AbstractGraphComponent, Name
-from b_asic.port import InputPort, OutputPort
-from b_asic.simulation import SimulationState, OperationState
-from b_asic.utilities import breadth_first_search
-from b_asic.signal import Signal
-
-
-class AbstractOperation(Operation, AbstractGraphComponent):
- """Generic abstract operation class which most implementations will derive from.
- TODO: More info.
- """
-
- _input_ports: List[InputPort]
- _output_ports: List[OutputPort]
- _parameters: Dict[str, Optional[Any]]
-
- def __init__(self, name: Name = ""):
- super().__init__(name)
- self._input_ports = []
- self._output_ports = []
- self._parameters = {}
-
- @abstractmethod
- def evaluate(self, *inputs) -> Any: # pylint: disable=arguments-differ
- """Evaluate the operation and generate a list of output values given a
- list of input values."""
- raise NotImplementedError
-
- def inputs(self) -> List[InputPort]:
- return self._input_ports.copy()
-
- def outputs(self) -> List[OutputPort]:
- return self._output_ports.copy()
-
- def input_count(self) -> int:
- return len(self._input_ports)
-
- def output_count(self) -> int:
- return len(self._output_ports)
-
- def input(self, i: int) -> InputPort:
- return self._input_ports[i]
-
- def output(self, i: int) -> OutputPort:
- return self._output_ports[i]
-
- def params(self) -> Dict[str, Optional[Any]]:
- return self._parameters.copy()
-
- def param(self, name: str) -> Optional[Any]:
- return self._parameters.get(name)
-
- def set_param(self, name: str, value: Any) -> None:
- assert name in self._parameters # TODO: Error message.
- self._parameters[name] = value
-
- def evaluate_outputs(self, state: SimulationState) -> List[Number]:
- # TODO: Check implementation.
- input_count: int = self.input_count()
- output_count: int = self.output_count()
- assert input_count == len(self._input_ports) # TODO: Error message.
- assert output_count == len(self._output_ports) # TODO: Error message.
-
- self_state: OperationState = state.operation_states[self]
-
- while self_state.iteration < state.iteration:
- input_values: List[Number] = [0] * input_count
- for i in range(input_count):
- 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)
- assert len(self_state.output_values) == output_count # TODO: Error message.
- self_state.iteration += 1
- for i in range(output_count):
- for signal in self._output_ports[i].signals():
- destination: Signal = signal.destination
- destination.evaluate_outputs(state)
-
- return self_state.output_values
-
- def split(self) -> List[Operation]:
- # TODO: Check implementation.
- results = self.evaluate(self._input_ports)
- if all(isinstance(e, Operation) for e in results):
- return results
- return [self]
-
- @property
- def neighbours(self) -> List[Operation]:
- neighbours: List[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
-
- def traverse(self) -> Operation:
- """Traverse the operation tree and return a generator with start point in the operation."""
- return breadth_first_search(self)
diff --git a/b_asic/core_operations.py b/b_asic/core_operations.py
index 7b644f34..ce1019f3 100644
--- a/b_asic/core_operations.py
+++ b/b_asic/core_operations.py
@@ -8,7 +8,7 @@ from typing import Any
from numpy import conjugate, sqrt, abs as np_abs
from b_asic.port import InputPort, OutputPort
from b_asic.graph_id import GraphIDType
-from b_asic.abstract_operation import AbstractOperation
+from b_asic.operation import AbstractOperation
from b_asic.graph_component import Name, TypeName
diff --git a/b_asic/operation.py b/b_asic/operation.py
index ac6e0319..75644b73 100644
--- a/b_asic/operation.py
+++ b/b_asic/operation.py
@@ -5,9 +5,12 @@ TODO: More info.
from abc import abstractmethod
from numbers import Number
-from typing import List, Dict, Optional, Any, TYPE_CHECKING
+from typing import List, Dict, Optional, Any, Set, TYPE_CHECKING
+from collections import deque
-from b_asic.graph_component import GraphComponent
+from b_asic.graph_component import GraphComponent, AbstractGraphComponent, Name
+from b_asic.simulation import SimulationState, OperationState
+from b_asic.signal import Signal
if TYPE_CHECKING:
from b_asic.port import InputPort, OutputPort
@@ -88,3 +91,114 @@ class Operation(GraphComponent):
If no neighbours are found this returns an empty list
"""
raise NotImplementedError
+
+
+class AbstractOperation(Operation, AbstractGraphComponent):
+ """Generic abstract operation class which most implementations will derive from.
+ TODO: More info.
+ """
+
+ _input_ports: List["InputPort"]
+ _output_ports: List["OutputPort"]
+ _parameters: Dict[str, Optional[Any]]
+
+ def __init__(self, name: Name = ""):
+ super().__init__(name)
+ self._input_ports = []
+ self._output_ports = []
+ self._parameters = {}
+
+ @abstractmethod
+ def evaluate(self, *inputs) -> Any: # pylint: disable=arguments-differ
+ """Evaluate the operation and generate a list of output values given a
+ list of input values."""
+ raise NotImplementedError
+
+ def inputs(self) -> List["InputPort"]:
+ return self._input_ports.copy()
+
+ def outputs(self) -> List["OutputPort"]:
+ return self._output_ports.copy()
+
+ def input_count(self) -> int:
+ return len(self._input_ports)
+
+ def output_count(self) -> int:
+ return len(self._output_ports)
+
+ def input(self, i: int) -> "InputPort":
+ return self._input_ports[i]
+
+ def output(self, i: int) -> "OutputPort":
+ return self._output_ports[i]
+
+ def params(self) -> Dict[str, Optional[Any]]:
+ return self._parameters.copy()
+
+ def param(self, name: str) -> Optional[Any]:
+ return self._parameters.get(name)
+
+ def set_param(self, name: str, value: Any) -> None:
+ assert name in self._parameters # TODO: Error message.
+ self._parameters[name] = value
+
+ def evaluate_outputs(self, state: SimulationState) -> List[Number]:
+ # TODO: Check implementation.
+ input_count: int = self.input_count()
+ output_count: int = self.output_count()
+ assert input_count == len(self._input_ports) # TODO: Error message.
+ assert output_count == len(self._output_ports) # TODO: Error message.
+
+ self_state: OperationState = state.operation_states[self]
+
+ while self_state.iteration < state.iteration:
+ input_values: List[Number] = [0] * input_count
+ for i in range(input_count):
+ 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)
+ assert len(self_state.output_values) == output_count # TODO: Error message.
+ self_state.iteration += 1
+ for i in range(output_count):
+ for signal in self._output_ports[i].signals():
+ destination: Signal = signal.destination
+ destination.evaluate_outputs(state)
+
+ return self_state.output_values
+
+ def split(self) -> List[Operation]:
+ # TODO: Check implementation.
+ results = self.evaluate(self._input_ports)
+ if all(isinstance(e, Operation) for e in results):
+ return results
+ return [self]
+
+ @property
+ def neighbours(self) -> List[Operation]:
+ neighbours: List[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
+
+ def traverse(self) -> Operation:
+ """Traverse the operation tree and return a generator with start point in the operation."""
+ return self._breadth_first_search()
+
+ def _breadth_first_search(self) -> Operation:
+ """Use breadth first search to traverse the operation tree."""
+ visited: Set[Operation] = {self}
+ queue = deque([self])
+ while queue:
+ operation = queue.popleft()
+ yield operation
+ for n_operation in operation.neighbours:
+ if n_operation not in visited:
+ visited.add(n_operation)
+ queue.append(n_operation)
diff --git a/b_asic/signal_flow_graph.py b/b_asic/signal_flow_graph.py
index f59a246d..9c08aecc 100644
--- a/b_asic/signal_flow_graph.py
+++ b/b_asic/signal_flow_graph.py
@@ -7,7 +7,7 @@ from typing import List, Dict, Optional, DefaultDict
from collections import defaultdict
from b_asic.operation import Operation
-from b_asic.abstract_operation import AbstractOperation
+from b_asic.operation import AbstractOperation
from b_asic.signal import Signal
from b_asic.graph_id import GraphIDGenerator, GraphID
from b_asic.graph_component import GraphComponent, Name, TypeName
diff --git a/b_asic/utilities.py b/b_asic/utilities.py
deleted file mode 100644
index 25707ff8..00000000
--- a/b_asic/utilities.py
+++ /dev/null
@@ -1,21 +0,0 @@
-"""@package docstring
-B-ASIC Operation Module.
-TODO: More info.
-"""
-
-from typing import Set
-from collections import deque
-
-from b_asic.operation import Operation
-
-def breadth_first_search(start: Operation) -> Operation:
- """Use breadth first search to traverse the operation tree."""
- visited: Set[Operation] = {start}
- queue = deque([start])
- while queue:
- operation = queue.popleft()
- yield operation
- for n_operation in operation.neighbours:
- if n_operation not in visited:
- visited.add(n_operation)
- queue.append(n_operation)
--
GitLab