Newer
Older
"""@package docstring
B-ASIC Operation Module.
TODO: More info.
"""
from abc import abstractmethod
from numbers import Number
Angus Lothian
committed
from typing import List, Dict, Optional, Any, Set, Sequence, TYPE_CHECKING
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
from collections import deque
from b_asic.graph_component import GraphComponent, AbstractGraphComponent, Name
if TYPE_CHECKING:
from b_asic.port import InputPort, OutputPort
class Operation(GraphComponent):
"""Operation interface.
TODO: More info.
"""
@abstractmethod
def inputs(self) -> "List[InputPort]":
"""Get a list of all input ports."""
raise NotImplementedError
@abstractmethod
def outputs(self) -> "List[OutputPort]":
"""Get a list of all output ports."""
raise NotImplementedError
@abstractmethod
def input_count(self) -> int:
"""Get the number of input ports."""
raise NotImplementedError
@abstractmethod
def output_count(self) -> int:
"""Get the number of output ports."""
raise NotImplementedError
@abstractmethod
def input(self, i: int) -> "InputPort":
"""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
Angus Lothian
committed
@abstractmethod
def evaluate_output(self, i: int, inputs: Sequence[Number]) -> Sequence[Optional[Number]]:
"""Evaluate the output port at the entered index with the entered input values and
returns all output values that are calulated during the evaluation in a list."""
raise NotImplementedError
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@abstractmethod
def params(self) -> Dict[str, Optional[Any]]:
"""Get a dictionary of all parameter values."""
raise NotImplementedError
@abstractmethod
def param(self, name: str) -> Optional[Any]:
"""Get the value of a parameter.
Returns None if the parameter is not defined.
"""
raise NotImplementedError
@abstractmethod
def set_param(self, name: str, value: Any) -> None:
"""Set the value of a parameter.
The parameter must be defined.
"""
raise NotImplementedError
@abstractmethod
def split(self) -> "List[Operation]":
"""Split the operation into multiple operations.
If splitting is not possible, this may return a list containing only the operation itself.
"""
raise NotImplementedError
@property
@abstractmethod
def neighbors(self) -> "List[Operation]":
"""Return all operations that are connected by signals to this operation.
If no neighbors 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
Angus Lothian
committed
def evaluate_output(self, i: int, inputs: Sequence[Number]) -> Sequence[Optional[Number]]:
eval_return = self.evaluate(*inputs)
if isinstance(eval_return, Number):
return [eval_return]
elif isinstance(eval_return, (list, tuple)):
return eval_return
else:
raise TypeError("Incorrect returned type from evaluate function.")
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
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 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 neighbors(self) -> List[Operation]:
neighbors: List[Operation] = []
for port in self._input_ports:
for signal in port.signals:
neighbors.append(signal.source.operation)
for port in self._output_ports:
for signal in port.signals:
neighbors.append(signal.destination.operation)
return neighbors
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.neighbors:
if n_operation not in visited:
visited.add(n_operation)
queue.append(n_operation)
def __add__(self, other):
"""Overloads the addition operator to make it return a new Addition operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantAddition operation object instead.
"""
# Import here to avoid circular imports.
from b_asic.core_operations import Addition, ConstantAddition
if isinstance(other, Operation):
return Addition(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantAddition(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")
def __sub__(self, other):
"""Overloads the subtraction operator to make it return a new Subtraction operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantSubtraction operation object instead.
"""
# Import here to avoid circular imports.
from b_asic.core_operations import Subtraction, ConstantSubtraction
if isinstance(other, Operation):
return Subtraction(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantSubtraction(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")
def __mul__(self, other):
"""Overloads the multiplication operator to make it return a new Multiplication operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantMultiplication operation object instead.
"""
# Import here to avoid circular imports.
from b_asic.core_operations import Multiplication, ConstantMultiplication
if isinstance(other, Operation):
return Multiplication(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantMultiplication(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")
def __truediv__(self, other):
"""Overloads the division operator to make it return a new Division operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantDivision operation object instead.
"""
# Import here to avoid circular imports.
from b_asic.core_operations import Division, ConstantDivision
if isinstance(other, Operation):
return Division(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantDivision(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")