Skip to content
Snippets Groups Projects
Commit 5fffc69b authored by Angus Lothian's avatar Angus Lothian :dark_sunglasses: Committed by Jacob Wahlman
Browse files

Change so that Basic Operation also inherits from abstract graph component and...

Change so that Basic Operation also inherits from abstract graph component and removed name property implementation
parent 68f9f1f7
No related branches found
No related tags found
3 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
......@@ -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 *
"""@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
"""@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
......
......@@ -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.
"""@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
......@@ -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
......@@ -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]
......
......@@ -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
......@@ -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"
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% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment