Skip to content
Snippets Groups Projects
Commit 96a9b53e authored by Ivar Härnqvist's avatar Ivar Härnqvist
Browse files

Merge branch 'pylint-tabs-to-spaces' into 'develop' 

Convert from tabs to spaces

See merge request PUM_TDDD96/B-ASIC!8
parents 3548e462 75204fba
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,!8Convert from tabs to spaces
Pipeline #10207 passed
Stage: test
Hide whitespace changes
Inline Side-by-side
Showing
with 470 additions and 471 deletions
b_asic/abstract_operation.py
+ 97
− 97
View file @ 96a9b53e
......@@ -15,100 +15,100 @@ from b_asic.utilities import breadth_first_search
from b_asic.abstract_graph_component import AbstractGraphComponent
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, **kwds):
super().__init__(**kwds)
self._input_ports = []
self._output_ports = []
self._parameters = {}
@abstractmethod
def evaluate(self, inputs: list) -> list:
"""Evaluate the operation and generate a list of output values given a list of input values."""
raise NotImplementedError
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 evaluate_outputs(self, state: SimulationState) -> List[Number]:
# TODO: Check implementation.
input_count: int = self.input_count()
output_count: int = self.output_count()
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]
while self_state.iteration < state.iteration:
input_values: List[Number] = [0] * input_count
for i in range(input_count):
source: Signal = self._input_ports[i].signal
input_values[i] = source.operation.evaluate_outputs(state)[source.port_index]
self_state.output_values = self.evaluate(input_values)
assert len(self_state.output_values) == output_count # TODO: Error message.
self_state.iteration += 1
for i in range(output_count):
for signal in self._output_ports[i].signals():
destination: Signal = signal.destination
destination.evaluate_outputs(state)
return self_state.output_values
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 neighbours(self) -> List[Operation]:
neighbours: List[Operation] = []
for port in self._input_ports:
for signal in port.signals:
neighbours.append(signal.source.operation)
for port in self._output_ports:
for signal in port.signals:
neighbours.append(signal.destination.operation)
return neighbours
def traverse(self) -> Operation:
"""Traverse the operation tree and return a generator with start point in the operation."""
return breadth_first_search(self)
# TODO: More stuff.
"""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, **kwds):
super().__init__(**kwds)
self._input_ports = []
self._output_ports = []
self._parameters = {}
@abstractmethod
def evaluate(self, inputs: list) -> list:
"""Evaluate the operation and generate a list of output values given a list of input values."""
raise NotImplementedError
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 evaluate_outputs(self, state: SimulationState) -> List[Number]:
# TODO: Check implementation.
input_count: int = self.input_count()
output_count: int = self.output_count()
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]
while self_state.iteration < state.iteration:
input_values: List[Number] = [0] * input_count
for i in range(input_count):
source: Signal = self._input_ports[i].signal
input_values[i] = source.operation.evaluate_outputs(state)[source.port_index]
self_state.output_values = self.evaluate(input_values)
assert len(self_state.output_values) == output_count # TODO: Error message.
self_state.iteration += 1
for i in range(output_count):
for signal in self._output_ports[i].signals():
destination: Signal = signal.destination
destination.evaluate_outputs(state)
return self_state.output_values
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 neighbours(self) -> List[Operation]:
neighbours: List[Operation] = []
for port in self._input_ports:
for signal in port.signals:
neighbours.append(signal.source.operation)
for port in self._output_ports:
for signal in port.signals:
neighbours.append(signal.destination.operation)
return neighbours
def traverse(self) -> Operation:
"""Traverse the operation tree and return a generator with start point in the operation."""
return breadth_first_search(self)
# TODO: More stuff.
b_asic/core_operations.py
+ 50
− 50
View file @ 96a9b53e
......@@ -13,70 +13,70 @@ from b_asic.graph_component import Name, TypeName
class Input(Operation, AbstractGraphComponent):
"""Input operation.
TODO: More info.
"""
"""Input operation.
TODO: More info.
"""
# TODO: Implement all functions.
# TODO: Implement all functions.
@property
def type_name(self) -> TypeName:
return "in"
@property
def type_name(self) -> TypeName:
return "in"
class Constant(AbstractOperation):
"""Constant value operation.
TODO: More info.
"""
"""Constant value operation.
TODO: More info.
"""
def __init__(self, value: Number, **kwds):
"""Construct a Constant."""
super().__init__(**kwds)
self._output_ports = [OutputPort(1, self)] # TODO: Generate appropriate ID for ports.
self._parameters["value"] = value
def __init__(self, value: Number, **kwds):
"""Construct a Constant."""
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 evaluate(self, inputs: list) -> list:
return [self.param("value")]
@property
def type_name(self) -> TypeName:
return "const"
@property
def type_name(self) -> TypeName:
return "const"
class Addition(AbstractOperation):
"""Binary addition operation.
TODO: More info.
"""
"""Binary addition operation.
TODO: More info.
"""
def __init__(self, **kwds):
"""Construct an Addition."""
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 __init__(self, **kwds):
"""Construct an Addition."""
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 evaluate(self, inputs: list) -> list:
return [inputs[0] + inputs[1]]
@property
def type_name(self) -> TypeName:
return "add"
@property
def type_name(self) -> TypeName:
return "add"
class ConstantMultiplication(AbstractOperation):
"""Unary constant multiplication operation.
TODO: More info.
"""
def __init__(self, coefficient: Number, **kwds):
"""Construct a ConstantMultiplication."""
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
def evaluate(self, inputs: list) -> list:
return [inputs[0] * self.param("coefficient")]
@property
def type_name(self) -> TypeName:
return "const_mul"
"""Unary constant multiplication operation.
TODO: More info.
"""
def __init__(self, coefficient: Number, **kwds):
"""Construct a ConstantMultiplication."""
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
def evaluate(self, inputs: list) -> list:
return [inputs[0] * self.param("coefficient")]
@property
def type_name(self) -> TypeName:
return "const_mul"
b_asic/operation.py
+ 78
− 78
View file @ 96a9b53e
......@@ -10,83 +10,83 @@ 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.port import InputPort, OutputPort
from b_asic.simulation import SimulationState
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
@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 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.
"""
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 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
"""
raise NotImplementedError
"""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
@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 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.
"""
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 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
"""
raise NotImplementedError
b_asic/port.py
+ 93
− 93
View file @ 96a9b53e
......@@ -13,117 +13,117 @@ PortId = NewType("PortId", int)
class Port(ABC):
"""Abstract port class.
TODO: More info.
"""
_port_id: PortId
_operation: Operation
def __init__(self, port_id: PortId, operation: Operation):
self._port_id = port_id
self._operation = operation
@property
def identifier(self) -> PortId:
"""Get the unique identifier."""
return self._port_id
@property
def operation(self) -> Operation:
"""Get the connected operation."""
return self._operation
@property
@abstractmethod
def signals(self) -> List[Signal]:
"""Get a list of all connected signals."""
raise NotImplementedError
@abstractmethod
def signal(self, i: int = 0) -> Signal:
"""Get the connected signal at index i."""
raise NotImplementedError
@abstractmethod
def signal_count(self) -> int:
"""Get the number of connected signals."""
raise NotImplementedError
@abstractmethod
def connect(self, signal: Signal) -> None:
"""Connect a signal."""
raise NotImplementedError
@abstractmethod
def disconnect(self, i: int = 0) -> None:
"""Disconnect a signal."""
raise NotImplementedError
"""Abstract port class.
TODO: More info.
"""
_port_id: PortId
_operation: Operation
def __init__(self, port_id: PortId, operation: Operation):
self._port_id = port_id
self._operation = operation
@property
def identifier(self) -> PortId:
"""Get the unique identifier."""
return self._port_id
@property
def operation(self) -> Operation:
"""Get the connected operation."""
return self._operation
@property
@abstractmethod
def signals(self) -> List[Signal]:
"""Get a list of all connected signals."""
raise NotImplementedError
@abstractmethod
def signal(self, i: int = 0) -> Signal:
"""Get the connected signal at index i."""
raise NotImplementedError
@abstractmethod
def signal_count(self) -> int:
"""Get the number of connected signals."""
raise NotImplementedError
@abstractmethod
def connect(self, signal: Signal) -> None:
"""Connect a signal."""
raise NotImplementedError
@abstractmethod
def disconnect(self, i: int = 0) -> None:
"""Disconnect a signal."""
raise NotImplementedError
class InputPort(Port):
"""Input port.
TODO: More info.
"""
_source_signal: Optional[Signal]
"""Input port.
TODO: More info.
"""
_source_signal: Optional[Signal]
def __init__(self, port_id: PortId, operation: Operation):
super().__init__(port_id, operation)
self._source_signal = None
def __init__(self, port_id: PortId, operation: Operation):
super().__init__(port_id, operation)
self._source_signal = None
@property
def signals(self) -> List[Signal]:
return [] if self._source_signal is None else [self._source_signal]
@property
def signals(self) -> List[Signal]:
return [] if self._source_signal is None else [self._source_signal]
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.
return self._source_signal
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.
return self._source_signal
def signal_count(self) -> int:
return 0 if self._source_signal is None else 1
def signal_count(self) -> int:
return 0 if self._source_signal is None else 1
def connect(self, signal: Signal) -> None:
self._source_signal = signal
signal.destination = self
def connect(self, signal: Signal) -> None:
self._source_signal = signal
signal.destination = self
def disconnect(self, i: int = 0) -> None:
assert 0 <= i < self.signal_count() # TODO: Error message.
self._source_signal.disconnect_source()
self._source_signal = None
def disconnect(self, i: int = 0) -> None:
assert 0 <= i < self.signal_count() # TODO: Error message.
self._source_signal.disconnect_source()
self._source_signal = None
# TODO: More stuff.
# TODO: More stuff.
class OutputPort(Port):
"""Output port.
TODO: More info.
"""
"""Output port.
TODO: More info.
"""
_destination_signals: List[Signal]
_destination_signals: List[Signal]
def __init__(self, port_id: PortId, operation: Operation):
super().__init__(port_id, operation)
self._destination_signals = []
def __init__(self, port_id: PortId, operation: Operation):
super().__init__(port_id, operation)
self._destination_signals = []
@property
def signals(self) -> List[Signal]:
return self._destination_signals.copy()
@property
def signals(self) -> List[Signal]:
return self._destination_signals.copy()
def signal(self, i: int = 0) -> Signal:
assert 0 <= i < self.signal_count() # TODO: Error message.
return self._destination_signals[i]
def signal(self, i: int = 0) -> Signal:
assert 0 <= i < self.signal_count() # TODO: Error message.
return self._destination_signals[i]
def signal_count(self) -> int:
return len(self._destination_signals)
def signal_count(self) -> int:
return len(self._destination_signals)
def connect(self, signal: Signal) -> None:
assert signal not in self._destination_signals # TODO: Error message.
self._destination_signals.append(signal)
signal.source = self
def connect(self, signal: Signal) -> None:
assert signal not in self._destination_signals # TODO: Error message.
self._destination_signals.append(signal)
signal.source = self
def disconnect(self, i: int = 0) -> None:
assert 0 <= i < self.signal_count() # TODO: Error message.
del self._destination_signals[i]
def disconnect(self, i: int = 0) -> None:
assert 0 <= i < self.signal_count() # TODO: Error message.
del self._destination_signals[i]
# TODO: More stuff.
# TODO: More stuff.
b_asic/precedence_chart.py
+ 9
− 9
View file @ 96a9b53e
......@@ -7,15 +7,15 @@ from b_asic.signal_flow_graph import SFG
class PrecedenceChart:
"""Precedence chart constructed from a signal flow graph.
TODO: More info.
"""
"""Precedence chart constructed from a signal flow graph.
TODO: More info.
"""
sfg: SFG
# TODO: More members.
sfg: SFG
# TODO: More members.
def __init__(self, sfg: SFG):
self.sfg = sfg
# TODO: Implement.
def __init__(self, sfg: SFG):
self.sfg = sfg
# TODO: Implement.
# TODO: More stuff.
# TODO: More stuff.
b_asic/schema.py
+ 9
− 9
View file @ 96a9b53e
......@@ -7,15 +7,15 @@ from b_asic.precedence_chart import PrecedenceChart
class Schema:
"""Schema constructed from a precedence chart.
TODO: More info.
"""
"""Schema constructed from a precedence chart.
TODO: More info.
"""
pc: PrecedenceChart
# TODO: More members.
pc: PrecedenceChart
# TODO: More members.
def __init__(self, pc: PrecedenceChart):
self.pc = pc
# TODO: Implement.
def __init__(self, pc: PrecedenceChart):
self.pc = pc
# TODO: Implement.
# TODO: More stuff.
# TODO: More stuff.
b_asic/signal.py
+ 41
− 41
View file @ 96a9b53e
......@@ -7,46 +7,46 @@ from b_asic.graph_component import TypeName
from b_asic.abstract_graph_component import AbstractGraphComponent
if TYPE_CHECKING:
from b_asic import OutputPort, InputPort
from b_asic import OutputPort, InputPort
class Signal(AbstractGraphComponent):
"""A connection between two ports."""
_source: "OutputPort"
_destination: "InputPort"
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) -> "OutputPort":
"""Returns the source OutputPort of the signal."""
return self._source
@property
def destination(self) -> "InputPort":
"""Returns the destination InputPort of the signal."""
return self._destination
@source.setter
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
"""A connection between two ports."""
_source: "OutputPort"
_destination: "InputPort"
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) -> "OutputPort":
"""Returns the source OutputPort of the signal."""
return self._source
@property
def destination(self) -> "InputPort":
"""Returns the destination InputPort of the signal."""
return self._destination
@source.setter
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
+ 75
− 75
View file @ 96a9b53e
......@@ -14,78 +14,78 @@ from b_asic.graph_component import GraphComponent, Name, TypeName
class SFG(AbstractOperation):
"""Signal flow graph.
TODO: More info.
"""
_graph_components_by_id: Dict[GraphID, GraphComponent]
_graph_components_by_name: DefaultDict[Name, List[GraphComponent]]
_graph_id_generator: GraphIDGenerator
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_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:
graph_component: Graph component to add to the graph.
"""
# Add to name dict
self._graph_components_by_name[graph_component.name].append(graph_component)
# 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[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_components_by_id:
return self._graph_components_by_id[graph_id]
return None
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"
"""Signal flow graph.
TODO: More info.
"""
_graph_components_by_id: Dict[GraphID, GraphComponent]
_graph_components_by_name: DefaultDict[Name, List[GraphComponent]]
_graph_id_generator: GraphIDGenerator
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_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:
graph_component: Graph component to add to the graph.
"""
# Add to name dict
self._graph_components_by_name[graph_component.name].append(graph_component)
# 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[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_components_by_id:
return self._graph_components_by_id[graph_id]
return None
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"
b_asic/simulation.py
+ 17
− 17
View file @ 96a9b53e
......@@ -8,28 +8,28 @@ from typing import List
class OperationState:
"""Simulation state of an operation.
TODO: More info.
"""
"""Simulation state of an operation.
TODO: More info.
"""
output_values: List[Number]
iteration: int
output_values: List[Number]
iteration: int
def __init__(self):
self.output_values = []
self.iteration = 0
def __init__(self):
self.output_values = []
self.iteration = 0
class SimulationState:
"""Simulation state.
TODO: More info.
"""
"""Simulation state.
TODO: More info.
"""
# operation_states: Dict[OperationId, OperationState]
iteration: int
# operation_states: Dict[OperationId, OperationState]
iteration: int
def __init__(self):
self.operation_states = {}
self.iteration = 0
def __init__(self):
self.operation_states = {}
self.iteration = 0
# TODO: More stuff.
# TODO: More stuff.
test/graph_id/test_graph_id_generator.py
+ 0
− 1
View file @ 96a9b53e
......@@ -26,4 +26,3 @@ def test_different_strings_generator():
assert graph_id_generator.get_next_id("mul") == "mul1"
assert graph_id_generator.get_next_id("sub") == "sub2"
assert graph_id_generator.get_next_id("mul") == "mul2"
\ No newline at end of file
test/port/test_outputport.py
+ 1
− 1
View file @ 96a9b53e
......@@ -15,4 +15,4 @@ def test_connect_multiple_signals(signals):
outp_port.connect(s)
assert outp_port.signal_count() == 3
assert outp_port.signals == signals
\ No newline at end of file
assert outp_port.signals == signals
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