Skip to content
Snippets Groups Projects
Commit 1541f59d authored by Jacob Wahlman's avatar Jacob Wahlman :ok_hand:
Browse files

Merge branch '3-operation-naming-system' into 'develop' 

Add implementation of #3, "Operation Naming System"

See merge request PUM_TDDD96/B-ASIC!6
parents 68f9f1f7 5fffc69b
No related branches found
No related tags found
4 merge requests!67WIP: B-ASIC version 1.0.0 hotfix,!65B-ASIC version 1.0.0,!15Add changes from sprint 1 and 2 to master,!6Add implementation of #3, "Operation Naming System"
Pipeline #10180 passed
Stage: test
Hide whitespace changes
Inline Side-by-side
Showing
with 197 additions and 97 deletions
b_asic/__init__.py
+ 5
− 2
View file @ 1541f59d
......@@ -2,9 +2,11 @@
Better ASIC Toolbox.
TODO: More info.
"""
from _b_asic import *
from b_asic.basic_operation import *
from b_asic.abstract_graph_component import *
from b_asic.abstract_operation import *
from b_asic.core_operations import *
from b_asic.graph_component import *
from b_asic.graph_id import *
from b_asic.operation import *
from b_asic.precedence_chart import *
from b_asic.port import *
......@@ -12,3 +14,4 @@ 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.traverse_tree import *
b_asic/abstract_graph_component.py 0 → 100644
+ 25
− 0
View file @ 1541f59d
"""@package docstring
B-ASIC module for Graph Components of a signal flow graph.
TODO: More info.
"""
from b_asic.graph_component import GraphComponent, Name
class AbstractGraphComponent(GraphComponent):
"""Abstract Graph Component class which is a component of a signal flow graph.
TODO: More info.
"""
_name: Name
def __init__(self, name: Name = ""):
self._name = name
@property
def name(self) -> Name:
return self._name
@name.setter
def name(self, name: Name) -> None:
self._name = name
b_asic/basic_operation.py → b_asic/abstract_operation.py
+ 7
− 7
View file @ 1541f59d
"""@package docstring
B-ASIC Basic Operation Module.
B-ASIC Abstract Operation Module.
TODO: More info.
"""
......@@ -11,9 +11,9 @@ from b_asic.port import InputPort, OutputPort
from b_asic.signal import Signal
from b_asic.operation import Operation
from b_asic.simulation import SimulationState, OperationState
from b_asic.abstract_graph_component import AbstractGraphComponent
class BasicOperation(Operation):
class AbstractOperation(Operation, AbstractGraphComponent):
"""Generic abstract operation class which most implementations will derive from.
TODO: More info.
"""
......@@ -22,8 +22,8 @@ class BasicOperation(Operation):
_output_ports: List[OutputPort]
_parameters: Dict[str, Optional[Any]]
def __init__(self):
"""Construct a BasicOperation."""
def __init__(self, **kwds):
super().__init__(**kwds)
self._input_ports = []
self._output_ports = []
self._parameters = {}
......@@ -31,7 +31,7 @@ class BasicOperation(Operation):
@abstractmethod
def evaluate(self, inputs: list) -> list:
"""Evaluate the operation and generate a list of output values given a list of input values."""
pass
raise NotImplementedError
def inputs(self) -> List[InputPort]:
return self._input_ports.copy()
......@@ -68,7 +68,7 @@ class BasicOperation(Operation):
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.identifier()]
self_state: OperationState = state.operation_states[self]
while self_state.iteration < state.iteration:
input_values: List[Number] = [0] * input_count
......
b_asic/core_operations.py
+ 24
− 18
View file @ 1541f59d
......@@ -7,62 +7,69 @@ from numbers import Number
from b_asic.port import InputPort, OutputPort
from b_asic.operation import Operation
from b_asic.basic_operation import BasicOperation
from b_asic.graph_id import GraphIDType
from b_asic.abstract_operation import AbstractOperation
from b_asic.abstract_graph_component import AbstractGraphComponent
from b_asic.graph_component import Name, TypeName
class Input(Operation):
class Input(Operation, AbstractGraphComponent):
"""Input operation.
TODO: More info.
"""
# TODO: Implement all functions.
pass
@property
def type_name(self) -> TypeName:
return "in"
class Constant(BasicOperation):
class Constant(AbstractOperation):
"""Constant value operation.
TODO: More info.
"""
def __init__(self, value: Number):
def __init__(self, value: Number, **kwds):
"""Construct a Constant."""
super().__init__()
super().__init__(**kwds)
self._output_ports = [OutputPort(1, self)] # TODO: Generate appropriate ID for ports.
self._parameters["value"] = value
def evaluate(self, inputs: list) -> list:
return [self.param("value")]
def type_name(self) -> GraphIDType:
@property
def type_name(self) -> TypeName:
return "const"
class Addition(BasicOperation):
class Addition(AbstractOperation):
"""Binary addition operation.
TODO: More info.
"""
def __init__(self):
def __init__(self, **kwds):
"""Construct an Addition."""
super().__init__()
super().__init__(**kwds)
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]]
def type_name(self) -> GraphIDType:
@property
def type_name(self) -> TypeName:
return "add"
class ConstantMultiplication(BasicOperation):
class ConstantMultiplication(AbstractOperation):
"""Unary constant multiplication operation.
TODO: More info.
"""
def __init__(self, coefficient: Number):
def __init__(self, coefficient: Number, **kwds):
"""Construct a ConstantMultiplication."""
super().__init__()
super().__init__(**kwds)
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
......@@ -70,7 +77,6 @@ class ConstantMultiplication(BasicOperation):
def evaluate(self, inputs: list) -> list:
return [inputs[0] * self.param("coefficient")]
def type_name(self) -> GraphIDType:
@property
def type_name(self) -> TypeName:
return "const_mul"
# TODO: More operations.
b_asic/graph_component.py 0 → 100644
+ 34
− 0
View file @ 1541f59d
"""@package docstring
B-ASIC Operation Module.
TODO: More info.
"""
from abc import ABC, abstractmethod
from typing import NewType
Name = NewType("Name", str)
TypeName = NewType("TypeName", str)
class GraphComponent(ABC):
"""Graph component interface.
TODO: More info.
"""
@property
@abstractmethod
def type_name(self) -> TypeName:
"""Returns the type name of the graph component"""
raise NotImplementedError
@property
@abstractmethod
def name(self) -> Name:
"""Returns the name of the graph component."""
raise NotImplementedError
@name.setter
@abstractmethod
def name(self, name: Name) -> None:
"""Sets the name of the graph component to the entered name."""
raise NotImplementedError
b_asic/operation.py
+ 19
− 22
View file @ 1541f59d
......@@ -3,17 +3,17 @@ B-ASIC Operation Module.
TODO: More info.
"""
from abc import ABC, abstractmethod
from abc import abstractmethod
from numbers import Number
from typing import List, Dict, Optional, Any, TYPE_CHECKING
from b_asic.graph_component import GraphComponent
if TYPE_CHECKING:
from b_asic.port import InputPort, OutputPort
from b_asic.simulation import SimulationState
from b_asic.graph_id import GraphIDType
class Operation(ABC):
class Operation(GraphComponent):
"""Operation interface.
TODO: More info.
"""
......@@ -21,75 +21,72 @@ class Operation(ABC):
@abstractmethod
def inputs(self) -> "List[InputPort]":
"""Get a list of all input ports."""
pass
raise NotImplementedError
@abstractmethod
def outputs(self) -> "List[OutputPort]":
"""Get a list of all output ports."""
pass
raise NotImplementedError
@abstractmethod
def input_count(self) -> int:
"""Get the number of input ports."""
pass
raise NotImplementedError
@abstractmethod
def output_count(self) -> int:
"""Get the number of output ports."""
pass
raise NotImplementedError
@abstractmethod
def input(self, i: int) -> "InputPort":
"""Get the input port at index i."""
pass
raise NotImplementedError
@abstractmethod
def output(self, i: int) -> "OutputPort":
"""Get the output port at index i."""
pass
raise NotImplementedError
@abstractmethod
def params(self) -> Dict[str, Optional[Any]]:
"""Get a dictionary of all parameter values."""
pass
raise NotImplementedError
@abstractmethod
def param(self, name: str) -> Optional[Any]:
"""Get the value of a parameter.
Returns None if the parameter is not defined.
"""
pass
raise NotImplementedError
@abstractmethod
def set_param(self, name: str, value: Any) -> None:
"""Set the value of a parameter.
The parameter must be defined.
"""
pass
raise NotImplementedError
@abstractmethod
def evaluate_outputs(self, state: "SimulationState") -> List[Number]:
"""Simulate the circuit until its iteration count matches that of the simulation state,
then return the resulting output vector.
"""
pass
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.
"""
pass
@abstractmethod
def type_name(self) -> "GraphIDType":
"""Returns a string representing the operation name of the operation."""
pass
raise NotImplementedError
@property
@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.
raise NotImplementedError
b_asic/port.py
+ 5
− 11
View file @ 1541f59d
......@@ -38,31 +38,27 @@ class Port(ABC):
@abstractmethod
def signals(self) -> List[Signal]:
"""Get a list of all connected signals."""
pass
raise NotImplementedError
@property
@abstractmethod
def signal(self, i: int = 0) -> Signal:
"""Get the connected signal at index i."""
pass
raise NotImplementedError
@abstractmethod
def signal_count(self) -> int:
"""Get the number of connected signals."""
pass
raise NotImplementedError
@abstractmethod
def connect(self, signal: Signal) -> None:
"""Connect a signal."""
pass
raise NotImplementedError
@abstractmethod
def disconnect(self, i: int = 0) -> None:
"""Disconnect a signal."""
pass
# TODO: More stuff.
raise NotImplementedError
class InputPort(Port):
......@@ -79,7 +75,6 @@ class InputPort(Port):
def signals(self) -> List[Signal]:
return [] if self._source_signal is None else [self._source_signal]
@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.
......@@ -115,7 +110,6 @@ class OutputPort(Port):
def signals(self) -> List[Signal]:
return self._destination_signals.copy()
@property
def signal(self, i: int = 0) -> Signal:
assert 0 <= i < self.signal_count() # TODO: Error message.
return self._destination_signals[i]
......
b_asic/signal.py
+ 21
− 7
View file @ 1541f59d
......@@ -2,37 +2,51 @@
B-ASIC Signal Module.
"""
from typing import TYPE_CHECKING, Optional
from b_asic.graph_component import TypeName
from b_asic.abstract_graph_component import AbstractGraphComponent
if TYPE_CHECKING:
from b_asic import OutputPort, InputPort
class Signal:
class Signal(AbstractGraphComponent):
"""A connection between two ports."""
_source: "OutputPort"
_destination: "InputPort"
def __init__(self, src: Optional["OutputPort"] = None, dest: Optional["InputPort"] = None):
def __init__(self, src: Optional["OutputPort"] = None, dest: Optional["InputPort"] = None, **kwds):
super().__init__(**kwds)
self._source = src
self._destination = dest
@property
def source(self) -> "InputPort":
@property
def source(self) -> "OutputPort":
"""Returns the source OutputPort of the signal."""
return self._source
@property
def destination(self) -> "OutputPort":
def destination(self) -> "InputPort":
"""Returns the destination InputPort of the signal."""
return self._destination
@source.setter
def source(self, src: "Outputport") -> None:
def source(self, src: "OutputPort") -> None:
"""Sets the value of the source OutputPort of the signal."""
self._source = src
@destination.setter
def destination(self, dest: "InputPort") -> None:
"""Sets the value of the destination InputPort of the signal."""
self._destination = dest
@property
def type_name(self) -> TypeName:
return "s"
def disconnect_source(self) -> None:
"""Disconnects the source OutputPort of the signal."""
self._source = None
def disconnect_destination(self) -> None:
"""Disconnects the destination InputPort of the signal."""
self._destination = None
b_asic/signal_flow_graph.py
+ 50
− 30
View file @ 1541f59d
......@@ -3,69 +3,89 @@ B-ASIC Signal Flow Graph Module.
TODO: More info.
"""
from typing import List, Dict, Union, Optional
from typing import List, Dict, Optional, DefaultDict
from collections import defaultdict
from b_asic.operation import Operation
from b_asic.basic_operation import BasicOperation
from b_asic.abstract_operation import AbstractOperation
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
from b_asic.graph_component import GraphComponent, Name, TypeName
class SFG(BasicOperation):
class SFG(AbstractOperation):
"""Signal flow graph.
TODO: More info.
"""
_graph_objects_by_id: Dict[GraphID, Union[Operation, Signal]]
_graph_components_by_id: Dict[GraphID, GraphComponent]
_graph_components_by_name: DefaultDict[Name, List[GraphComponent]]
_graph_id_generator: GraphIDGenerator
def __init__(self, input_destinations: List[Signal], output_sources: List[Signal]):
super().__init__()
# TODO: Allocate input/output ports with appropriate IDs.
self._graph_objects_by_id = dict # Map Operation ID to Operation objects
def __init__(self, input_signals: List[Signal] = None, output_signals: List[Signal] = None, \
ops: List[Operation] = None, **kwds):
super().__init__(**kwds)
if input_signals is None:
input_signals = []
if output_signals is None:
output_signals = []
if ops is None:
ops = []
self._graph_components_by_id = dict() # Maps Graph ID to objects
self._graph_components_by_name = defaultdict(list) # Maps Name to objects
self._graph_id_generator = GraphIDGenerator()
for operation in ops:
self._add_graph_component(operation)
for input_signal in input_signals:
self._add_graph_component(input_signal)
# TODO: Construct SFG based on what inputs that were given
# 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 add_operation(self, operation: Operation) -> GraphID:
"""Adds the entered operation to the SFG's dictionary of graph objects and
def _add_graph_component(self, graph_component: GraphComponent) -> GraphID:
"""Adds the entered graph component to the SFG's dictionary of graph objects and
returns a generated GraphID for it.
Keyword arguments:
operation: Operation to add to the graph.
graph_component: Graph component to add to the graph.
"""
return self._add_graph_obj(operation, operation.type_name())
# Add to name dict
self._graph_components_by_name[graph_component.name].append(graph_component)
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')
# Add to ID dict
graph_id: GraphID = self._graph_id_generator.get_next_id(graph_component.type_name)
self._graph_components_by_id[graph_id] = graph_component
return graph_id
def find_by_id(self, graph_id: GraphID) -> Optional[Operation]:
def find_by_id(self, graph_id: GraphID) -> Optional[GraphComponent]:
"""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]
if graph_id in self._graph_components_by_id:
return self._graph_components_by_id[graph_id]
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
def find_by_name(self, name: Name) -> List[GraphComponent]:
"""Finds all graph objects that have the entered name and returns them
in a list. If no graph object with the entered name was found then returns an
empty list.
Keyword arguments:
name: Name of the wanted object.
"""
return self._graph_components_by_name[name]
@property
def type_name(self) -> TypeName:
return "sfg"
test/signal_flow_graph/test_signal_flow_graph.py
+ 7
− 0
View file @ 1541f59d
from b_asic.signal_flow_graph import SFG
from b_asic.core_operations import Addition, Constant
from b_asic.signal import Signal
from b_asic.signal_flow_graph import SFG
import pytest
def test_adding_to_sfg():
pass
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment