From 5fffc69bcaae6b81ba570e2295f40914c24c5c6e Mon Sep 17 00:00:00 2001
From: Angus Lothian <anglo547@student.liu.se>
Date: Thu, 5 Mar 2020 10:57:04 +0100
Subject: [PATCH] Change so that Basic Operation also inherits from abstract
graph component and removed name property implementation
---
b_asic/__init__.py | 7 +-
b_asic/abstract_graph_component.py | 25 ++++++
...sic_operation.py => abstract_operation.py} | 14 ++--
b_asic/core_operations.py | 42 +++++-----
b_asic/graph_component.py | 34 ++++++++
b_asic/operation.py | 41 +++++-----
b_asic/port.py | 16 ++--
b_asic/signal.py | 28 +++++--
b_asic/signal_flow_graph.py | 80 ++++++++++++-------
.../test_signal_flow_graph.py | 7 ++
10 files changed, 197 insertions(+), 97 deletions(-)
create mode 100644 b_asic/abstract_graph_component.py
rename b_asic/{basic_operation.py => abstract_operation.py} (90%)
create mode 100644 b_asic/graph_component.py
diff --git a/b_asic/__init__.py b/b_asic/__init__.py
index 752bac07..4ae6652b 100644
--- a/b_asic/__init__.py
+++ b/b_asic/__init__.py
@@ -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 *
diff --git a/b_asic/abstract_graph_component.py b/b_asic/abstract_graph_component.py
new file mode 100644
index 00000000..6efb94e3
--- /dev/null
+++ b/b_asic/abstract_graph_component.py
@@ -0,0 +1,25 @@
+"""@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
diff --git a/b_asic/basic_operation.py b/b_asic/abstract_operation.py
similarity index 90%
rename from b_asic/basic_operation.py
rename to b_asic/abstract_operation.py
index 93a27222..ab8438a5 100644
--- a/b_asic/basic_operation.py
+++ b/b_asic/abstract_operation.py
@@ -1,5 +1,5 @@
"""@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
diff --git a/b_asic/core_operations.py b/b_asic/core_operations.py
index 45919b8b..52f18361 100644
--- a/b_asic/core_operations.py
+++ b/b_asic/core_operations.py
@@ -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.
diff --git a/b_asic/graph_component.py b/b_asic/graph_component.py
new file mode 100644
index 00000000..4bce63f2
--- /dev/null
+++ b/b_asic/graph_component.py
@@ -0,0 +1,34 @@
+"""@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
diff --git a/b_asic/operation.py b/b_asic/operation.py
index 923690aa..5d4b404d 100644
--- a/b_asic/operation.py
+++ b/b_asic/operation.py
@@ -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
diff --git a/b_asic/port.py b/b_asic/port.py
index 4c6fb244..a8a062fc 100644
--- a/b_asic/port.py
+++ b/b_asic/port.py
@@ -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]
diff --git a/b_asic/signal.py b/b_asic/signal.py
index 17078138..810c00dc 100644
--- a/b_asic/signal.py
+++ b/b_asic/signal.py
@@ -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
-
diff --git a/b_asic/signal_flow_graph.py b/b_asic/signal_flow_graph.py
index f7d4be64..38c46697 100644
--- a/b_asic/signal_flow_graph.py
+++ b/b_asic/signal_flow_graph.py
@@ -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"
diff --git a/test/signal_flow_graph/test_signal_flow_graph.py b/test/signal_flow_graph/test_signal_flow_graph.py
index 921e8906..d18d2da5 100644
--- a/test/signal_flow_graph/test_signal_flow_graph.py
+++ b/test/signal_flow_graph/test_signal_flow_graph.py
@@ -1,3 +1,10 @@
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
+
--
GitLab