Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • da/B-ASIC
  • lukja239/B-ASIC
  • robal695/B-ASIC
3 results
Show changes
Commits on Source (2)
Hide whitespace changes
Inline Side-by-side
Showing
with 554 additions and 157 deletions
README.md
View file @ df381f3e
......@@ -29,6 +29,15 @@ To run the test suite, the following additional packages are required:
* pytest
* pytest-cov (for testing with coverage)
To build a binary distribution, the following additional packages are required:
* Python:
* wheel
To run the test suite, the following additional packages are required:
* Python:
* pytest
* pytest-cov (for testing with coverage)
### Using CMake directly
How to build using CMake.
......
b_asic/__init__.py
View file @ df381f3e
......@@ -10,3 +10,4 @@ from b_asic.signal_flow_graph import *
from b_asic.signal import *
from b_asic.simulation import *
from b_asic.special_operations import *
from b_asic.schema import *
b_asic/core_operations.py
View file @ df381f3e
......@@ -4,7 +4,7 @@ TODO: More info.
"""
from numbers import Number
from typing import Optional
from typing import Optional, Dict
from numpy import conjugate, sqrt, abs as np_abs
from b_asic.port import SignalSourceProvider, InputPort, OutputPort
......@@ -44,9 +44,9 @@ class Addition(AbstractOperation):
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count=2, output_count=1,
name=name, input_sources=[src0, src1])
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
super().__init__(input_count=2, output_count=1, name=name, input_sources=[src0, src1],
latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
......@@ -61,9 +61,9 @@ class Subtraction(AbstractOperation):
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count=2, output_count=1,
name=name, input_sources=[src0, src1])
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
super().__init__(input_count=2, output_count=1, name=name, input_sources=[src0, src1],
latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
......@@ -78,9 +78,9 @@ class Multiplication(AbstractOperation):
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count=2, output_count=1,
name=name, input_sources=[src0, src1])
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
super().__init__(input_count=2, output_count=1, name=name, input_sources=[src0, src1],
latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
......@@ -95,9 +95,9 @@ class Division(AbstractOperation):
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count=2, output_count=1,
name=name, input_sources=[src0, src1])
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
super().__init__(input_count=2, output_count=1, name=name, input_sources=[src0, src1],
latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
......@@ -112,9 +112,9 @@ class Min(AbstractOperation):
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count=2, output_count=1,
name=name, input_sources=[src0, src1])
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
super().__init__(input_count=2, output_count=1, name=name, input_sources=[src0, src1],
latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
......@@ -131,9 +131,9 @@ class Max(AbstractOperation):
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count=2, output_count=1,
name=name, input_sources=[src0, src1])
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
super().__init__(input_count=2, output_count=1, name=name, input_sources=[src0, src1],
latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
......@@ -150,9 +150,9 @@ class SquareRoot(AbstractOperation):
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count=1, output_count=1,
name=name, input_sources=[src0])
def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
super().__init__(input_count=1, output_count=1, name=name, input_sources=[src0],
latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
......@@ -167,9 +167,9 @@ class ComplexConjugate(AbstractOperation):
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count=1, output_count=1,
name=name, input_sources=[src0])
def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
super().__init__(input_count=1, output_count=1, name=name, input_sources=[src0],
latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
......@@ -184,9 +184,9 @@ class Absolute(AbstractOperation):
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count=1, output_count=1,
name=name, input_sources=[src0])
def __init__(self, src0: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
super().__init__(input_count=1, output_count=1, name=name, input_sources=[src0],
latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
......@@ -201,9 +201,9 @@ class ConstantMultiplication(AbstractOperation):
TODO: More info.
"""
def __init__(self, value: Number = 0, src0: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count=1, output_count=1,
name=name, input_sources=[src0])
def __init__(self, value: Number = 0, src0: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
super().__init__(input_count=1, output_count=1, name=name, input_sources=[src0],
latency=latency, latency_offsets=latency_offsets)
self.set_param("value", value)
@classmethod
......@@ -230,9 +230,9 @@ class Butterfly(AbstractOperation):
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count=2, output_count=2,
name=name, input_sources=[src0, src1])
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
super().__init__(input_count=2, output_count=2, name=name, input_sources=[src0, src1],
latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
......@@ -241,13 +241,15 @@ class Butterfly(AbstractOperation):
def evaluate(self, a, b):
return a + b, a - b
class MAD(AbstractOperation):
"""Multiply-and-add operation.
TODO: More info.
"""
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, src2: Optional[SignalSourceProvider] = None, name: Name = ""):
super().__init__(input_count = 3, output_count = 1, name = name, input_sources = [src0, src1, src2])
def __init__(self, src0: Optional[SignalSourceProvider] = None, src1: Optional[SignalSourceProvider] = None, src2: Optional[SignalSourceProvider] = None, name: Name = "", latency: int = None, latency_offsets: Dict[str, int] = None):
super().__init__(input_count=3, output_count=1, name=name, input_sources=[src0, src1, src2],
latency=latency, latency_offsets=latency_offsets)
@classmethod
def type_name(cls) -> TypeName:
......
b_asic/operation.py
View file @ df381f3e
......@@ -3,16 +3,17 @@ B-ASIC Operation Module.
TODO: More info.
"""
from b_asic.signal import Signal
from b_asic.port import SignalSourceProvider, InputPort, OutputPort
from b_asic.graph_component import GraphComponent, AbstractGraphComponent, Name
import itertools as it
from math import trunc
import collections
from abc import abstractmethod
from numbers import Number
from typing import NewType, List, Sequence, Iterable, Mapping, MutableMapping, Optional, Any, Set, Union
from math import trunc
from typing import NewType, List, Dict, Sequence, Iterable, Mapping, MutableMapping, Optional, Any, Set, Union
from b_asic.graph_component import GraphComponent, AbstractGraphComponent, Name
from b_asic.port import SignalSourceProvider, InputPort, OutputPort
from b_asic.signal import Signal
OutputKey = NewType("OutputKey", str)
OutputMap = Mapping[OutputKey, Optional[Number]]
......@@ -193,6 +194,39 @@ class Operation(GraphComponent, SignalSourceProvider):
"""
raise NotImplementedError
@property
@abstractmethod
def latency(self) -> int:
"""Get the latency of the operation, which is the longest time it takes from one of
the operations inputport to one of the operations outputport.
"""
raise NotImplementedError
@property
@abstractmethod
def latency_offsets(self) -> Sequence[Sequence[int]]:
"""Get a nested list with all the operations ports latency-offsets, the first list contains the
latency-offsets of the operations input ports, the second list contains the latency-offsets of
the operations output ports.
"""
raise NotImplementedError
@abstractmethod
def set_latency(self, latency: int) -> None:
"""Sets the latency of the operation to the specified integer value by setting the
latency-offsets of operations input ports to 0 and the latency-offsets of the operations
output ports to the specified value. The latency cannot be a negative integers.
"""
raise NotImplementedError
@abstractmethod
def set_latency_offsets(self, latency_offsets: Dict[str, int]) -> None:
"""Sets the latency-offsets for the operations ports specified in the latency_offsets dictionary.
The latency offsets dictionary should be {'in0': 2, 'out1': 4} if you want to set the latency offset
for the inport port with index 0 to 2, and the latency offset of the output port with index 1 to 4.
"""
raise NotImplementedError
class AbstractOperation(Operation, AbstractGraphComponent):
"""Generic abstract operation class which most implementations will derive from.
......@@ -202,7 +236,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
_input_ports: List[InputPort]
_output_ports: List[OutputPort]
def __init__(self, input_count: int, output_count: int, name: Name = "", input_sources: Optional[Sequence[Optional[SignalSourceProvider]]] = None):
def __init__(self, input_count: int, output_count: int, name: Name = "", input_sources: Optional[Sequence[Optional[SignalSourceProvider]]] = None, latency: int = None, latency_offsets: Dict[str, int] = None):
super().__init__(name)
self._input_ports = [InputPort(self, i) for i in range(input_count)]
......@@ -218,6 +252,22 @@ class AbstractOperation(Operation, AbstractGraphComponent):
if src is not None:
self._input_ports[i].connect(src.source)
ports_without_latency_offset = set(([f"in{i}" for i in range(self.input_count)] +
[f"out{i}" for i in range(self.output_count)]))
if latency_offsets is not None:
self.set_latency_offsets(latency_offsets)
if latency is not None:
# Set the latency of the rest of ports with no latency_offset.
assert latency >= 0, "Negative latency entered"
for inp in self.inputs:
if inp.latency_offset is None:
inp.latency_offset = 0
for outp in self.outputs:
if outp.latency_offset is None:
outp.latency_offset = latency
@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."""
......@@ -433,6 +483,14 @@ class AbstractOperation(Operation, AbstractGraphComponent):
return SFG(inputs=inputs, outputs=outputs)
def copy_component(self, *args, **kwargs) -> Operation:
new_component = super().copy_component(*args, **kwargs)
for i, inp in enumerate(self.inputs):
new_component.input(i).latency_offset = inp.latency_offset
for i, outp in enumerate(self.outputs):
new_component.output(i).latency_offset = outp.latency_offset
return new_component
def inputs_required_for_output(self, output_index: int) -> Iterable[int]:
if output_index < 0 or output_index >= self.output_count:
raise IndexError(f"Output index out of range (expected 0-{self.output_count - 1}, got {output_index})")
......@@ -483,3 +541,32 @@ class AbstractOperation(Operation, AbstractGraphComponent):
else:
args.append(input_values[i])
return args
@property
def latency(self) -> int:
return max(((outp.latency_offset - inp.latency_offset) for outp, inp in it.product(self.outputs, self.inputs)))
def set_latency(self, latency: int) -> None:
assert latency >= 0, "Negative latency entered."
for inport in self.inputs:
inport.latency_offset = 0
for outport in self.outputs:
outport.latency_offset = latency
@property
def latency_offsets(self) -> Sequence[Sequence[int]]:
return ([inp.latency_offset for inp in self.inputs], [outp.latency_offset for outp in self.outputs])
def set_latency_offsets(self, latency_offsets: Dict[str, int]) -> None:
for port_str, latency_offset in latency_offsets.items():
port_str = port_str.lower()
if port_str.startswith("in"):
index_str = port_str[2:]
assert index_str.isdigit(), "Incorrectly formatted index in string, expected 'in' + index"
self.input(int(index_str)).latency_offset = latency_offset
elif port_str.startswith("out"):
index_str = port_str[3:]
assert index_str.isdigit(), "INcorrectly formatted index in string, expected 'out' + index"
self.output(int(index_str)).latency_offset = latency_offset
else:
raise ValueError("Incorrectly formatted string, expected 'in' + index or 'out' + index")
b_asic/port.py
View file @ df381f3e
......@@ -32,6 +32,18 @@ class Port(ABC):
"""Return the index of the port."""
raise NotImplementedError
@property
@abstractmethod
def latency_offset(self) -> int:
"""Get the latency_offset of the port."""
raise NotImplementedError
@latency_offset.setter
@abstractmethod
def latency_offset(self, latency_offset: int) -> None:
"""Set the latency_offset of the port to the integer specified value."""
raise NotImplementedError
@property
@abstractmethod
def signal_count(self) -> int:
......@@ -76,10 +88,12 @@ class AbstractPort(Port):
_operation: "Operation"
_index: int
_latency_offset: Optional[int]
def __init__(self, operation: "Operation", index: int):
def __init__(self, operation: "Operation", index: int, latency_offset: int = None):
self._operation = operation
self._index = index
self._latency_offset = latency_offset
@property
def operation(self) -> "Operation":
......@@ -89,6 +103,14 @@ class AbstractPort(Port):
def index(self) -> int:
return self._index
@property
def latency_offset(self) -> int:
return self._latency_offset
@latency_offset.setter
def latency_offset(self, latency_offset: int):
self._latency_offset = latency_offset
class SignalSourceProvider(ABC):
"""Signal source provider interface.
......
b_asic/schema.py 0 → 100644
View file @ df381f3e
"""@package docstring
This module contains the Schema class.
TODO: More info
"""
from typing import Dict, List
from b_asic.signal_flow_graph import SFG
from b_asic.graph_component import GraphID
from b_asic.operation import Operation
class Schema:
"""A class that represents an SFG with scheduled Operations."""
_sfg: SFG
_start_times: Dict[GraphID, int]
_laps: Dict[GraphID, List[int]]
_schedule_time: int
_cyclic: bool
_resolution: int
def __init__(self, sfg: SFG, schedule_time: int = None, cyclic: bool = False, resolution: int = 1, scheduling_alg: str = "ASAP"):
self._sfg = sfg
self._start_times = dict()
self._laps = dict()
self._cyclic = cyclic
self._resolution = resolution
if scheduling_alg == "ASAP":
self._schedule_asap()
else:
raise NotImplementedError(f"No algorithm with name: {scheduling_alg} defined.")
max_end_time = 0
for op_id, op_start_time in self._start_times.items():
op = self._sfg.find_by_id(op_id)
for outport in op.outputs:
max_end_time = max(max_end_time, op_start_time + outport.latency_offset)
if not self._cyclic:
if schedule_time is None:
self._schedule_time = max_end_time
elif schedule_time < max_end_time:
raise ValueError("Too short schedule time for non-cyclic Schedule entered.")
else:
self._schedule_time = schedule_time
def start_time_of_operation(self, op_id: GraphID):
"""Get the start time of the operation with the specified by the op_id."""
assert op_id in self._start_times, "No operation with the specified op_id in this schema."
return self._start_times[op_id]
def forward_slack(self, op_id):
raise NotImplementedError
def backward_slack(self, op_id):
raise NotImplementedError
def print_slacks(self):
raise NotImplementedError
def _schedule_asap(self):
pl = self._sfg.get_precedence_list()
if len(pl) < 2:
print("Empty signal flow graph cannot be scheduled.")
return
non_schedulable_ops = set((outp.operation.graph_id for outp in pl[0]))
for outport in pl[1]:
op = outport.operation
if op not in self._start_times:
# Set start time of all operations in the first iter to 0
self._start_times[op.graph_id] = 0
for outports in pl[2:]:
for outport in outports:
op = outport.operation
if op.graph_id not in self._start_times:
# Schedule the operation if it doesn't have a start time yet.
op_start_time = 0
for inport in op.inputs:
print(inport.operation.graph_id)
assert len(inport.signals) == 1, "Error in scheduling, dangling input port detected."
assert inport.signals[0].source is not None, "Error in scheduling, signal with no source detected."
source_port = inport.signals[0].source
source_end_time = None
if source_port.operation.graph_id in non_schedulable_ops:
source_end_time = 0
else:
source_op_time = self._start_times[source_port.operation.graph_id]
assert source_port.latency_offset is not None, f"Output port: {source_port.index} of operation: \
{source_port.operation.graph_id} has no latency-offset."
assert inport.latency_offset is not None, f"Input port: {inport.index} of operation: \
{inport.operation.graph_id} has no latency-offset."
source_end_time = source_op_time + source_port.latency_offset
op_start_time_from_in = source_end_time - inport.latency_offset
op_start_time = max(op_start_time, op_start_time_from_in)
self._start_times[op.graph_id] = op_start_time
b_asic/signal_flow_graph.py
View file @ df381f3e
......@@ -8,7 +8,7 @@ from numbers import Number
from collections import defaultdict, deque
from io import StringIO
from queue import PriorityQueue
import itertools
import itertools as it
from graphviz import Digraph
from b_asic.port import SignalSourceProvider, OutputPort
......@@ -612,7 +612,7 @@ class SFG(AbstractOperation):
def show_precedence_graph(self) -> None:
p_list = self.get_precedence_list()
pg = Digraph()
pg.attr(rankdir = 'LR')
pg.attr(rankdir='LR')
# Creates nodes for each output port in the precedence list
for i in range(len(p_list)):
......@@ -627,11 +627,11 @@ class SFG(AbstractOperation):
for port in ports:
for signal in port.signals:
pg.edge(port.operation.graph_id + '.' + str(port.index), signal.destination.operation.graph_id)
pg.node(signal.destination.operation.graph_id, shape = 'square')
pg.node(signal.destination.operation.graph_id, shape='square')
pg.edge(port.operation.graph_id, port.operation.graph_id + '.' + str(port.index))
pg.node(port.operation.graph_id, shape = 'square')
pg.node(port.operation.graph_id, shape='square')
pg.view()
pg.view()
def print_precedence_graph(self) -> None:
"""Prints a representation of the SFG's precedence list to the standard out.
......@@ -682,14 +682,18 @@ class SFG(AbstractOperation):
first_op = no_inputs_queue.popleft()
visited = set([first_op])
p_queue = PriorityQueue()
p_queue.put((-first_op.output_count, first_op)) # Negative priority as max-heap popping is wanted
p_queue_entry_num = it.count()
# Negative priority as max-heap popping is wanted
p_queue.put((-first_op.output_count, -next(p_queue_entry_num), first_op))
operations_left = len(self.operations) - 1
seen_but_not_visited_count = 0
while operations_left > 0:
while not p_queue.empty():
op = p_queue.get()[1]
op = p_queue.get()[2]
operations_left -= 1
top_order.append(op)
......@@ -701,7 +705,7 @@ class SFG(AbstractOperation):
remaining_inports = remaining_inports_per_operation[neighbor_op]
if remaining_inports == 0:
p_queue.put((-neighbor_op.output_count, neighbor_op))
p_queue.put((-neighbor_op.output_count, -next(p_queue_entry_num), neighbor_op))
elif remaining_inports > 0:
if neighbor_op in seen:
......@@ -717,15 +721,15 @@ class SFG(AbstractOperation):
# First check if can fetch from Operations with no input ports
if no_inputs_queue:
new_op = no_inputs_queue.popleft()
p_queue.put((new_op.output_count, new_op))
p_queue.put((-new_op.output_count, -next(p_queue_entry_num), new_op))
# Else fetch operation with lowest input count that is not zero
elif seen_but_not_visited_count > 0:
for i in itertools.count(start=1):
for i in it.count(start=1):
seen_inputs_queue = seen_with_inputs_dict[i]
if seen_inputs_queue:
new_op = seen_inputs_queue.popleft()
p_queue.put((-new_op.output_count, new_op))
p_queue.put((-new_op.output_count, -next(p_queue_entry_num), new_op))
seen_but_not_visited_count -= 1
break
else:
......@@ -734,3 +738,11 @@ class SFG(AbstractOperation):
self._operations_topological_order = top_order
return self._operations_topological_order
def set_latency_of_type(self, type_name: TypeName, latency: int):
for op in self.get_components_with_type_name(type_name):
op.set_latency(latency)
def set_latency_offsets_of_type(self, type_name: TypeName, latency_offsets: Dict[str, int]):
for op in self.get_components_with_type_name(type_name):
op.set_latency_offsets(latency_offsets)
b_asic/special_operations.py
View file @ df381f3e
......@@ -6,7 +6,7 @@ TODO: More info.
from numbers import Number
from typing import Optional, Sequence
from b_asic.operation import AbstractOperation, OutputKey, RegisterMap, MutableOutputMap, MutableRegisterMap
from b_asic.operation import AbstractOperation, RegisterMap, MutableOutputMap, MutableRegisterMap
from b_asic.graph_component import Name, TypeName
from b_asic.port import SignalSourceProvider
......
test/fixtures/signal_flow_graph.py
View file @ df381f3e
......@@ -21,12 +21,12 @@ def sfg_two_inputs_two_outputs():
out1 = in1 + in2
out2 = in1 + 2 * in2
"""
in1 = Input()
in2 = Input()
add1 = in1 + in2
add2 = add1 + in2
out1 = Output(add1)
out2 = Output(add2)
in1 = Input("IN1")
in2 = Input("IN2")
add1 = Addition(in1, in2, "ADD1")
add2 = Addition(add1, in2, "ADD2")
out1 = Output(add1, "OUT1")
out2 = Output(add2, "OUT2")
return SFG(inputs=[in1, in2], outputs=[out1, out2])
......@@ -58,6 +58,31 @@ def sfg_two_inputs_two_outputs_independent():
return SFG(inputs=[in1, in2], outputs=[out1, out2])
@pytest.fixture
def sfg_two_inputs_two_outputs_independent_with_cmul():
"""Valid SFG with two inputs and two outputs, where the first output only depends
on the first input and the second output only depends on the second input.
. .
in1--->cmul1--->cmul2--->out1
. .
. .
c1------+ .
|
v .
in2--->add1---->cmul3--->out2
"""
in1 = Input("IN1")
in2 = Input("IN2")
c1 = Constant(3, "C1")
add1 = Addition(in2, c1, "ADD1", 7)
cmul3 = ConstantMultiplication(2, add1, "CMUL3", 3)
cmul1 = ConstantMultiplication(5, in1, "CMUL1", 5)
cmul2 = ConstantMultiplication(4, cmul1, "CMUL2", 4)
out1 = Output(in1, "OUT1")
out2 = Output(add1, "OUT2")
return SFG(inputs=[in1, in2], outputs=[out1, out2])
@pytest.fixture
def sfg_nested():
"""Valid SFG with two inputs and one output.
......
test/test_abstract_operation.py deleted 100644 → 0
View file @ e90e40fa
"""
B-ASIC test suite for the AbstractOperation class.
"""
import pytest
from b_asic import Addition, Subtraction, Multiplication, ConstantMultiplication, Division
def test_addition_overload():
"""Tests addition overloading for both operation and number argument."""
add1 = Addition(None, None, "add1")
add2 = Addition(None, None, "add2")
add3 = add1 + add2
assert isinstance(add3, Addition)
assert add3.input(0).signals == add1.output(0).signals
assert add3.input(1).signals == add2.output(0).signals
add4 = add3 + 5
assert isinstance(add4, Addition)
assert add4.input(0).signals == add3.output(0).signals
assert add4.input(1).signals[0].source.operation.value == 5
add5 = 5 + add4
assert isinstance(add5, Addition)
assert add5.input(0).signals[0].source.operation.value == 5
assert add5.input(1).signals == add4.output(0).signals
def test_subtraction_overload():
"""Tests subtraction overloading for both operation and number argument."""
add1 = Addition(None, None, "add1")
add2 = Addition(None, None, "add2")
sub1 = add1 - add2
assert isinstance(sub1, Subtraction)
assert sub1.input(0).signals == add1.output(0).signals
assert sub1.input(1).signals == add2.output(0).signals
sub2 = sub1 - 5
assert isinstance(sub2, Subtraction)
assert sub2.input(0).signals == sub1.output(0).signals
assert sub2.input(1).signals[0].source.operation.value == 5
sub3 = 5 - sub2
assert isinstance(sub3, Subtraction)
assert sub3.input(0).signals[0].source.operation.value == 5
assert sub3.input(1).signals == sub2.output(0).signals
def test_multiplication_overload():
"""Tests multiplication overloading for both operation and number argument."""
add1 = Addition(None, None, "add1")
add2 = Addition(None, None, "add2")
mul1 = add1 * add2
assert isinstance(mul1, Multiplication)
assert mul1.input(0).signals == add1.output(0).signals
assert mul1.input(1).signals == add2.output(0).signals
mul2 = mul1 * 5
assert isinstance(mul2, ConstantMultiplication)
assert mul2.input(0).signals == mul1.output(0).signals
assert mul2.value == 5
mul3 = 5 * mul2
assert isinstance(mul3, ConstantMultiplication)
assert mul3.input(0).signals == mul2.output(0).signals
assert mul3.value == 5
def test_division_overload():
"""Tests division overloading for both operation and number argument."""
add1 = Addition(None, None, "add1")
add2 = Addition(None, None, "add2")
div1 = add1 / add2
assert isinstance(div1, Division)
assert div1.input(0).signals == add1.output(0).signals
assert div1.input(1).signals == add2.output(0).signals
div2 = div1 / 5
assert isinstance(div2, Division)
assert div2.input(0).signals == div1.output(0).signals
assert div2.input(1).signals[0].source.operation.value == 5
div3 = 5 / div2
assert isinstance(div3, Division)
assert div3.input(0).signals[0].source.operation.value == 5
assert div3.input(1).signals == div2.output(0).signals
test/test_operation.py
View file @ df381f3e
"""
B-ASIC test suite for the AbstractOperation class.
"""
import pytest
from b_asic import Constant, Addition, MAD, Butterfly, SquareRoot
from b_asic import Addition, Subtraction, Multiplication, ConstantMultiplication, Division, Constant, Butterfly, \
MAD, SquareRoot
class TestOperationOverloading:
def test_addition_overload(self):
"""Tests addition overloading for both operation and number argument."""
add1 = Addition(None, None, "add1")
add2 = Addition(None, None, "add2")
add3 = add1 + add2
assert isinstance(add3, Addition)
assert add3.input(0).signals == add1.output(0).signals
assert add3.input(1).signals == add2.output(0).signals
add4 = add3 + 5
assert isinstance(add4, Addition)
assert add4.input(0).signals == add3.output(0).signals
assert add4.input(1).signals[0].source.operation.value == 5
add5 = 5 + add4
assert isinstance(add5, Addition)
assert add5.input(0).signals[0].source.operation.value == 5
assert add5.input(1).signals == add4.output(0).signals
def test_subtraction_overload(self):
"""Tests subtraction overloading for both operation and number argument."""
add1 = Addition(None, None, "add1")
add2 = Addition(None, None, "add2")
sub1 = add1 - add2
assert isinstance(sub1, Subtraction)
assert sub1.input(0).signals == add1.output(0).signals
assert sub1.input(1).signals == add2.output(0).signals
sub2 = sub1 - 5
assert isinstance(sub2, Subtraction)
assert sub2.input(0).signals == sub1.output(0).signals
assert sub2.input(1).signals[0].source.operation.value == 5
sub3 = 5 - sub2
assert isinstance(sub3, Subtraction)
assert sub3.input(0).signals[0].source.operation.value == 5
assert sub3.input(1).signals == sub2.output(0).signals
def test_multiplication_overload(self):
"""Tests multiplication overloading for both operation and number argument."""
add1 = Addition(None, None, "add1")
add2 = Addition(None, None, "add2")
mul1 = add1 * add2
assert isinstance(mul1, Multiplication)
assert mul1.input(0).signals == add1.output(0).signals
assert mul1.input(1).signals == add2.output(0).signals
mul2 = mul1 * 5
assert isinstance(mul2, ConstantMultiplication)
assert mul2.input(0).signals == mul1.output(0).signals
assert mul2.value == 5
mul3 = 5 * mul2
assert isinstance(mul3, ConstantMultiplication)
assert mul3.input(0).signals == mul2.output(0).signals
assert mul3.value == 5
def test_division_overload(self):
"""Tests division overloading for both operation and number argument."""
add1 = Addition(None, None, "add1")
add2 = Addition(None, None, "add2")
div1 = add1 / add2
assert isinstance(div1, Division)
assert div1.input(0).signals == add1.output(0).signals
assert div1.input(1).signals == add2.output(0).signals
div2 = div1 / 5
assert isinstance(div2, Division)
assert div2.input(0).signals == div1.output(0).signals
assert div2.input(1).signals[0].source.operation.value == 5
div3 = 5 / div2
assert isinstance(div3, Division)
assert div3.input(0).signals[0].source.operation.value == 5
assert div3.input(1).signals == div2.output(0).signals
class TestTraverse:
def test_traverse_single_tree(self, operation):
......@@ -24,20 +112,21 @@ class TestTraverse:
def test_traverse_loop(self, operation_graph_with_cycle):
assert len(list(operation_graph_with_cycle.traverse())) == 8
class TestToSfg:
def test_convert_mad_to_sfg(self):
mad1 = MAD()
mad1_sfg = mad1.to_sfg()
assert mad1.evaluate(1,1,1) == mad1_sfg.evaluate(1,1,1)
assert mad1.evaluate(1, 1, 1) == mad1_sfg.evaluate(1, 1, 1)
assert len(mad1_sfg.operations) == 6
def test_butterfly_to_sfg(self):
but1 = Butterfly()
but1_sfg = but1.to_sfg()
assert but1.evaluate(1,1)[0] == but1_sfg.evaluate(1,1)[0]
assert but1.evaluate(1,1)[1] == but1_sfg.evaluate(1,1)[1]
assert but1.evaluate(1, 1)[0] == but1_sfg.evaluate(1, 1)[0]
assert but1.evaluate(1, 1)[1] == but1_sfg.evaluate(1, 1)[1]
assert len(but1_sfg.operations) == 8
def test_add_to_sfg(self):
......@@ -51,3 +140,47 @@ class TestToSfg:
sqrt1_sfg = sqrt1.to_sfg()
assert len(sqrt1_sfg.operations) == 3
class TestLatency:
def test_latency_constructor(self):
bfly = Butterfly(latency=5)
assert bfly.latency == 5
assert list(bfly.latency_offsets) == [[0, 0], [5, 5]]
def test_latency_offsets_constructor(self):
bfly = Butterfly(latency_offsets={'in0': 2, 'in1': 3, 'out0': 5, 'out1': 10})
assert bfly.latency == 8
assert list(bfly.latency_offsets) == [[2, 3], [5, 10]]
def test_latency_and_latency_offsets_constructor(self):
bfly = Butterfly(latency=5, latency_offsets={'in1': 2, 'out0': 9})
assert bfly.latency == 9
assert list(bfly.latency_offsets) == [[0, 2], [9, 5]]
def test_set_latency(self):
bfly = Butterfly()
bfly.set_latency(9)
assert bfly.latency == 9
assert list(bfly.latency_offsets) == [[0, 0], [9, 9]]
def test_set_latency_offsets(self):
bfly = Butterfly()
bfly.set_latency_offsets({'in0': 3, 'out1': 5})
assert list(bfly.latency_offsets) == [[3, None], [None, 5]]
class TestCopyOperation:
def test_copy_buttefly_latency_offsets(self):
bfly = Butterfly(latency_offsets={'in0': 4, 'in1': 2, 'out0': 10, 'out1': 9})
bfly_copy = bfly.copy_component()
assert list(bfly_copy.latency_offsets) == [[4, 2], [10, 9]]
test/test_schema.py 0 → 100644
View file @ df381f3e
"""
B-ASIC test suite for the schema module and Schema class.
"""
from b_asic import Schema, Addition, ConstantMultiplication
class TestInit:
def test_simple_filter_normal_latency(self, simple_filter):
simple_filter.set_latency_of_type(Addition.type_name(), 5)
simple_filter.set_latency_of_type(ConstantMultiplication.type_name(), 4)
schema = Schema(simple_filter)
assert schema._start_times == {"add1": 4, "cmul1": 0}
def test_complicated_single_outputs_normal_latency(self, precedence_sfg_registers):
precedence_sfg_registers.set_latency_of_type(Addition.type_name(), 4)
precedence_sfg_registers.set_latency_of_type(ConstantMultiplication.type_name(), 3)
schema = Schema(precedence_sfg_registers, scheduling_alg="ASAP")
for op in schema._sfg.get_operations_topological_order():
print(op.latency_offsets)
start_times_names = dict()
for op_id, start_time in schema._start_times.items():
op_name = precedence_sfg_registers.find_by_id(op_id).name
start_times_names[op_name] = start_time
assert start_times_names == {"C0": 0, "B1": 0, "B2": 0, "ADD2": 3, "ADD1": 7, "Q1": 11,
"A0": 14, "A1": 0, "A2": 0, "ADD3": 3, "ADD4": 17}
def test_complicated_single_outputs_complex_latencies(self, precedence_sfg_registers):
precedence_sfg_registers.set_latency_offsets_of_type(ConstantMultiplication.type_name(), {'in0': 3, 'out0': 5})
precedence_sfg_registers.find_by_name("B1")[0].set_latency_offsets({'in0': 4, 'out0': 7})
precedence_sfg_registers.find_by_name("B2")[0].set_latency_offsets({'in0': 1, 'out0': 4})
precedence_sfg_registers.find_by_name("ADD2")[0].set_latency_offsets({'in0': 4, 'in1': 2, 'out0': 4})
precedence_sfg_registers.find_by_name("ADD1")[0].set_latency_offsets({'in0': 1, 'in1': 2, 'out0': 4})
precedence_sfg_registers.find_by_name("Q1")[0].set_latency_offsets({'in0': 3, 'out0': 6})
precedence_sfg_registers.find_by_name("A0")[0].set_latency_offsets({'in0': 0, 'out0': 2})
precedence_sfg_registers.find_by_name("A1")[0].set_latency_offsets({'in0': 0, 'out0': 5})
precedence_sfg_registers.find_by_name("A2")[0].set_latency_offsets({'in0': 2, 'out0': 3})
precedence_sfg_registers.find_by_name("ADD3")[0].set_latency_offsets({'in0': 2, 'in1': 1, 'out0': 4})
precedence_sfg_registers.find_by_name("ADD4")[0].set_latency_offsets({'in0': 6, 'in1': 7, 'out0': 9})
schema = Schema(precedence_sfg_registers, scheduling_alg="ASAP")
start_times_names = dict()
for op_id, start_time in schema._start_times.items():
op_name = precedence_sfg_registers.find_by_id(op_id).name
start_times_names[op_name] = start_time
assert start_times_names == {'C0': 0, 'B1': 0, 'B2': 0, 'ADD2': 3, 'ADD1': 5, 'Q1': 6, 'A0': 12,
'A1': 0, 'A2': 0, 'ADD3': 3, 'ADD4': 8}
def test_independent_sfg(self, sfg_two_inputs_two_outputs_independent_with_cmul):
schema = Schema(sfg_two_inputs_two_outputs_independent_with_cmul, scheduling_alg="ASAP")
start_times_names = dict()
for op_id, start_time in schema._start_times.items():
op_name = sfg_two_inputs_two_outputs_independent_with_cmul.find_by_id(op_id).name
start_times_names[op_name] = start_time
assert start_times_names == {'CMUL1': 0, 'CMUL2': 5, "ADD1": 0, "CMUL3": 7}
test/test_sfg.py
View file @ df381f3e
......@@ -691,12 +691,13 @@ class TestConnectExternalSignalsToComponentsMultipleComp:
out1.input(0).connect(sub1, "S7")
test_sfg = SFG(inputs=[inp1, inp2, inp3, inp4], outputs=[out1])
assert test_sfg.evaluate(1, 2, 3, 4) == 16
sfg1.connect_external_signals_to_components()
assert test_sfg.evaluate(1, 2, 3, 4) == 16
assert not test_sfg.connect_external_signals_to_components()
class TestTopologicalOrderOperations:
def test_feedback_sfg(self, simple_filter):
topological_order = simple_filter.get_operations_topological_order()
......@@ -708,6 +709,12 @@ class TestTopologicalOrderOperations:
assert [comp.name for comp in topological_order] == ["IN1", "OUT1", "IN2", "C1", "ADD1", "OUT2"]
def test_complex_graph(self, precedence_sfg_registers):
topological_order = precedence_sfg_registers.get_operations_topological_order()
assert [comp.name for comp in topological_order] == \
['IN1', 'C0', 'ADD1', 'Q1', 'A0', 'T1', 'B1', 'A1', 'T2', 'B2', 'ADD2', 'A2', 'ADD3', 'ADD4', 'OUT1']
class TestRemove:
def test_remove_single_input_outputs(self, simple_filter):
......@@ -776,3 +783,19 @@ class TestRemove:
def remove_different_number_inputs_outputs(self, simple_filter):
with pytest.raises(ValueError):
simple_filter.remove_operation("add1")
class TestGetComponentsOfType:
def test_get_no_operations_of_type(self, sfg_two_inputs_two_outputs):
assert [op.name for op in sfg_two_inputs_two_outputs.get_components_with_type_name(Multiplication.type_name())] \
== []
def test_get_multple_operations_of_type(self, sfg_two_inputs_two_outputs):
assert [op.name for op in sfg_two_inputs_two_outputs.get_components_with_type_name(Addition.type_name())] \
== ["ADD1", "ADD2"]
assert [op.name for op in sfg_two_inputs_two_outputs.get_components_with_type_name(Input.type_name())] \
== ["IN1", "IN2"]
assert [op.name for op in sfg_two_inputs_two_outputs.get_components_with_type_name(Output.type_name())] \
== ["OUT1", "OUT2"]