diff --git a/b_asic/abstract_operation.py b/b_asic/abstract_operation.py
index 558022a7b12304c2052d3263f20c9d2339ad8531..7cee1790930ce6de9dd056696a07153f020f8671 100644
--- a/b_asic/abstract_operation.py
+++ b/b_asic/abstract_operation.py
@@ -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.
diff --git a/b_asic/core_operations.py b/b_asic/core_operations.py
index 52f18361de75c1f8bc7ea415a8501916298796c9..f023e1a53417abb6b4613e3af48e2460ed9fa038 100644
--- a/b_asic/core_operations.py
+++ b/b_asic/core_operations.py
@@ -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"
diff --git a/b_asic/operation.py b/b_asic/operation.py
index 5d4b404ddcb4f2f8925a9808bcaec4712bb49c12..4a716b7900958a7eabae188bf763513afa5ab5e5 100644
--- a/b_asic/operation.py
+++ b/b_asic/operation.py
@@ -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
diff --git a/b_asic/port.py b/b_asic/port.py
index a8a062fc2f6b30a255aff44b49c8026a7d735dab..6cbd59ba21806567b76899cd98dbdc40ad4ddf18 100644
--- a/b_asic/port.py
+++ b/b_asic/port.py
@@ -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.
diff --git a/b_asic/precedence_chart.py b/b_asic/precedence_chart.py
index 93b86164fec041c20d9b170839897ecff96ccfdf..be55a123e0ab4330057c0bb62581e45195f5e5ba 100644
--- a/b_asic/precedence_chart.py
+++ b/b_asic/precedence_chart.py
@@ -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.
diff --git a/b_asic/schema.py b/b_asic/schema.py
index 41938263d144a066822befc5ad0a3a2ab41839c4..e5068cdc080c5c5004c44c885ac48f52ba44c1f3 100644
--- a/b_asic/schema.py
+++ b/b_asic/schema.py
@@ -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.
diff --git a/b_asic/signal.py b/b_asic/signal.py
index 810c00dc48a3d366f264851b10ab0a58569b2951..4d80530e5442e0413839673dfd3685e81568a088 100644
--- a/b_asic/signal.py
+++ b/b_asic/signal.py
@@ -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
diff --git a/b_asic/signal_flow_graph.py b/b_asic/signal_flow_graph.py
index 38c46697dcde8f701945dc27ca11965909b68336..6a91521afa50cd0261b2eff0b78f149170316ad4 100644
--- a/b_asic/signal_flow_graph.py
+++ b/b_asic/signal_flow_graph.py
@@ -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"
diff --git a/b_asic/simulation.py b/b_asic/simulation.py
index c4f7f8f366a5298ab7104b14741d61a3cc42f7c9..50adaa522b6d685b428354a9f84689330b7fd40f 100644
--- a/b_asic/simulation.py
+++ b/b_asic/simulation.py
@@ -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.
diff --git a/test/graph_id/test_graph_id_generator.py b/test/graph_id/test_graph_id_generator.py
index 7aeb6cad27e43233a88eb69e58bd89f78a863c5b..8af36e8c1713c92a7790d30039019176ebe3bc4f 100644
--- a/test/graph_id/test_graph_id_generator.py
+++ b/test/graph_id/test_graph_id_generator.py
@@ -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
diff --git a/test/port/test_outputport.py b/test/port/test_outputport.py
index 5c76bb480fa63488073f6dab9f82c3f3ce00b4f3..6ca126d549de8d73051731a31225063501da8013 100644
--- a/test/port/test_outputport.py
+++ b/test/port/test_outputport.py
@@ -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