Skip to content
Snippets Groups Projects
Commit 90b77b93 authored by Oscar Gustafsson's avatar Oscar Gustafsson :bicyclist:
Browse files

More documentation fixes

parent 0643a143
No related branches found
No related tags found
No related merge requests found
Pipeline #87767 failed
"""B-ASIC Core Operations Module.
"""
B-ASIC Core Operations Module.
Contains some of the most commonly used mathematical operations.
"""
......@@ -49,7 +50,7 @@ class Constant(AbstractOperation):
@value.setter
def value(self, value: Number) -> None:
"""Set the constant value of this operation."""
return self.set_param("value", value)
self.set_param("value", value)
class Addition(AbstractOperation):
......@@ -164,7 +165,7 @@ class AddSub(AbstractOperation):
@is_add.setter
def is_add(self, is_add: bool) -> None:
"""Set if operation is add."""
return self.set_param("is_add", is_add)
self.set_param("is_add", is_add)
class Multiplication(AbstractOperation):
......@@ -453,7 +454,7 @@ class ConstantMultiplication(AbstractOperation):
@value.setter
def value(self, value: Number) -> None:
"""Set the constant value of this operation."""
return self.set_param("value", value)
self.set_param("value", value)
class Butterfly(AbstractOperation):
......@@ -571,4 +572,4 @@ class SymmetricTwoportAdaptor(AbstractOperation):
@value.setter
def value(self, value: Number) -> None:
"""Set the constant value of this operation."""
return self.set_param("value", value)
self.set_param("value", value)
"""B-ASIC Graph Component Module.
"""
B-ASIC Graph Component Module.
Contains the base for all components with an ID in a signal flow graph.
"""
......
......@@ -38,7 +38,7 @@ class Operation(GraphComponent, SignalSourceProvider):
"""Operation interface.
Operations are graph components that perform a certain function.
They are connected to eachother by signals through their input/output
They are connected to each other by signals through their input/output
ports.
Operations can be evaluated independently using evaluate_output().
......@@ -82,7 +82,7 @@ class Operation(GraphComponent, SignalSourceProvider):
self, src: Union[SignalSourceProvider, Number]
) -> "Union[Multiplication, ConstantMultiplication]":
"""Overloads the multiplication operator to make it return a new Multiplication operation
object that is connected to the self and other objects. If other is a number then
object that is connected to the self and other objects. If *src* is a number, then
returns a ConstantMultiplication operation object instead.
"""
raise NotImplementedError
......@@ -91,8 +91,9 @@ class Operation(GraphComponent, SignalSourceProvider):
def __rmul__(
self, src: Union[SignalSourceProvider, Number]
) -> "Union[Multiplication, ConstantMultiplication]":
"""Overloads the multiplication operator to make it return a new Multiplication operation
object that is connected to the self and other objects. If other is a number then
"""
Overloads the multiplication operator to make it return a new Multiplication operation
object that is connected to the self and other objects. If *src* is a number, then
returns a ConstantMultiplication operation object instead.
"""
raise NotImplementedError
......@@ -101,7 +102,8 @@ class Operation(GraphComponent, SignalSourceProvider):
def __truediv__(
self, src: Union[SignalSourceProvider, Number]
) -> "Division":
"""Overloads the division operator to make it return a new Division operation
"""
Overloads the division operator to make it return a new Division operation
object that is connected to the self and other objects.
"""
raise NotImplementedError
......@@ -110,14 +112,16 @@ class Operation(GraphComponent, SignalSourceProvider):
def __rtruediv__(
self, src: Union[SignalSourceProvider, Number]
) -> "Division":
"""Overloads the division operator to make it return a new Division operation
"""
Overloads the division operator to make it return a new Division operation
object that is connected to the self and other objects.
"""
raise NotImplementedError
@abstractmethod
def __lshift__(self, src: SignalSourceProvider) -> Signal:
"""Overloads the left shift operator to make it connect the provided signal source
"""
Overloads the left shift operator to make it connect the provided signal source
to this operation's input, assuming it has exactly 1 input port.
Returns the new signal.
"""
......@@ -160,7 +164,8 @@ class Operation(GraphComponent, SignalSourceProvider):
@property
@abstractmethod
def input_signals(self) -> Iterable[Signal]:
"""Get all the signals that are connected to this operation's input ports,
"""
Get all the signals that are connected to this operation's input ports,
in no particular order.
"""
raise NotImplementedError
......@@ -168,14 +173,16 @@ class Operation(GraphComponent, SignalSourceProvider):
@property
@abstractmethod
def output_signals(self) -> Iterable[Signal]:
"""Get all the signals that are connected to this operation's output ports,
"""
Get all the signals that are connected to this operation's output ports,
in no particular order.
"""
raise NotImplementedError
@abstractmethod
def key(self, index: int, prefix: str = "") -> ResultKey:
"""Get the key used to access the output of a certain output of this operation
"""
Get the key used to access the output of a certain output of this operation
from the output parameter passed to current_output(s) or evaluate_output(s).
"""
raise NotImplementedError
......@@ -184,7 +191,8 @@ class Operation(GraphComponent, SignalSourceProvider):
def current_output(
self, index: int, delays: Optional[DelayMap] = None, prefix: str = ""
) -> Optional[Number]:
"""Get the current output at the given index of this operation, if available.
"""
Get the current output at the given index of this operation, if available.
The delays parameter will be used for lookup.
The prefix parameter will be used as a prefix for the key string when looking for delays.
See also: current_outputs, evaluate_output, evaluate_outputs.
......@@ -212,7 +220,7 @@ class Operation(GraphComponent, SignalSourceProvider):
The bits_override parameter specifies a word length override when truncating inputs
which ignores the word length specified by the input signal.
The truncate parameter specifies whether input truncation should be enabled in the first
place. If set to False, input values will be used driectly without any bit truncation.
place. If set to False, input values will be used directly without any bit truncation.
See also: evaluate_outputs, current_output, current_outputs.
"""
raise NotImplementedError
......@@ -257,20 +265,24 @@ class Operation(GraphComponent, SignalSourceProvider):
@abstractmethod
def inputs_required_for_output(self, output_index: int) -> Iterable[int]:
"""Get the input indices of all inputs in this operation whose values are required in order to evaluate the output at the given output index.
"""
Get the input indices of all inputs in this operation whose values are
required in order to evaluate the output at the given output index.
"""
raise NotImplementedError
@abstractmethod
def truncate_input(self, index: int, value: Number, bits: int) -> Number:
"""Truncate the value to be used as input at the given index to a certain bit length.
"""
Truncate the value to be used as input at the given index to a certain bit length.
"""
raise NotImplementedError
@property
@abstractmethod
def latency(self) -> int:
"""Get the latency of the operation, which is the longest time it takes from one of
"""
Get the latency of the operation, which is the longest time it takes from one of
the operations inputport to one of the operations outputport.
"""
raise NotImplementedError
......@@ -294,7 +306,8 @@ class Operation(GraphComponent, SignalSourceProvider):
@abstractmethod
def set_latency_offsets(self, latency_offsets: Dict[str, int]) -> None:
"""Sets the latency-offsets for the operations ports specified in the latency_offsets dictionary.
"""
Sets the latency-offsets for the operations ports specified in the latency_offsets dictionary.
The latency offsets dictionary should be {'in0': 2, 'out1': 4} if you want to set the latency offset
for the inport port with index 0 to 2, and the latency offset of the output port with index 1 to 4.
"""
......@@ -405,6 +418,8 @@ class AbstractOperation(Operation, AbstractGraphComponent):
if outp.latency_offset is None:
outp.latency_offset = latency
self._execution_time = execution_time
@abstractmethod
def evaluate(self, *inputs) -> Any: # pylint: disable=arguments-differ
"""Evaluate the operation and generate a list of output values given a list of input values.
......@@ -709,7 +724,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
from b_asic.signal_flow_graph import SFG
from b_asic.special_operations import Input, Output
inputs = [Input() for i in range(self.input_count)]
inputs = [Input() for _ in range(self.input_count)]
try:
last_operations = self.evaluate(*inputs)
......@@ -788,7 +803,9 @@ class AbstractOperation(Operation, AbstractGraphComponent):
input_values: Sequence[Number],
bits_override: Optional[int] = None,
) -> Sequence[Number]:
"""Truncate the values to be used as inputs to the bit lengths specified by the respective signals connected to each input.
"""
Truncate the values to be used as inputs to the bit lengths specified
by the respective signals connected to each input.
"""
args = []
for i, input_port in enumerate(self.inputs):
......@@ -938,4 +955,4 @@ class AbstractOperation(Operation, AbstractGraphComponent):
]
for k in range(len(self.outputs))
]
return (input_coords, output_coords)
return input_coords, output_coords
......@@ -147,7 +147,8 @@ class SignalSourceProvider(ABC):
class InputPort(AbstractPort):
"""Input port.
"""
Input port.
May have one or zero signals connected to it.
"""
......@@ -190,7 +191,8 @@ class InputPort(AbstractPort):
@property
def connected_source(self) -> Optional["OutputPort"]:
"""Get the output port that is currently connected to this input port,
"""
Get the output port that is currently connected to this input port,
or None if it is unconnected.
"""
return (
......@@ -198,7 +200,8 @@ class InputPort(AbstractPort):
)
def connect(self, src: SignalSourceProvider, name: Name = "") -> Signal:
"""Connect the provided signal source to this input port by creating a new signal.
"""
Connect the provided signal source to this input port by creating a new signal.
Returns the new signal.
"""
assert (
......@@ -208,14 +211,16 @@ class InputPort(AbstractPort):
return Signal(source=src.source, destination=self, name=name)
def __lshift__(self, src: SignalSourceProvider) -> Signal:
"""Overloads the left shift operator to make it connect the provided signal source to this input port.
Returns the new signal.
"""
Overloads the left shift operator to make it connect the provided
signal source to this input port. Returns the new signal.
"""
return self.connect(src)
class OutputPort(AbstractPort, SignalSourceProvider):
"""Output port.
"""
Output port.
May have zero or more signals connected to it.
"""
......
......@@ -80,8 +80,8 @@ class Signal(AbstractGraphComponent):
connect to the entered destination InputPort. Also connect the entered
destination port to the signal if it hasn't already been connected.
Keywords argments:
- dest: InputPort to connect as destination to the signal.
Keywords arguments:
- dest : InputPort to connect as destination to the signal.
"""
if dest is not self._destination:
self.remove_destination()
......
......@@ -158,7 +158,7 @@ class SFG(AbstractOperation):
self._input_operations.append(new_input_op)
self._original_input_signals_to_indices[signal] = input_index
# Setup input operations, starting from indices ater input signals.
# Setup input operations, starting from indices after input signals.
if inputs is not None:
for input_index, input_op in enumerate(inputs, input_signal_count):
assert (
......@@ -204,7 +204,6 @@ class SFG(AbstractOperation):
), "Duplicate output operations supplied to SFG constructor."
new_output_op = self._add_component_unconnected_copy(output_op)
for signal in output_op.input(0).signals:
new_signal = None
if signal in self._original_components_to_new:
# Signal was already added when setting up inputs.
new_signal = self._original_components_to_new[signal]
......@@ -390,13 +389,15 @@ class SFG(AbstractOperation):
"""
if len(self.inputs) != len(self.input_operations):
raise IndexError(
f"Number of inputs does not match the number of"
f" input_operations in SFG."
f"Number of inputs ({len(self.inputs)}) does not match the"
f" number of input_operations ({len(self.input_operations)})"
" in SFG."
)
if len(self.outputs) != len(self.output_operations):
raise IndexError(
f"Number of outputs does not match the number of"
f" output_operations SFG."
f"Number of outputs ({len(self.outputs)}) does not match the"
f" number of output_operations ({len(self.output_operations)})"
" in SFG."
)
if len(self.input_signals) == 0:
return False
......@@ -419,13 +420,15 @@ class SFG(AbstractOperation):
@property
def input_operations(self) -> Sequence[Operation]:
"""Get the internal input operations in the same order as their respective input ports.
"""
Get the internal input operations in the same order as their respective input ports.
"""
return self._input_operations
@property
def output_operations(self) -> Sequence[Operation]:
"""Get the internal output operations in the same order as their respective output ports.
"""
Get the internal output operations in the same order as their respective output ports.
"""
return self._output_operations
......@@ -498,11 +501,14 @@ class SFG(AbstractOperation):
def find_by_type_name(
self, type_name: TypeName
) -> Sequence[GraphComponent]:
"""Find all components in this graph with the specified type name.
"""
Find all components in this graph with the specified type name.
Returns an empty sequence if no components were found.
Keyword arguments:
type_name: The type_name of the desired components.
Parameters
==========
type_name : TypeName
The TypeName of the desired components.
"""
reg = f"{type_name}[0-9]+"
p = re.compile(reg)
......@@ -515,8 +521,11 @@ class SFG(AbstractOperation):
"""Find the graph component with the specified ID.
Returns None if the component was not found.
Keyword arguments:
graph_id: Graph ID of the desired component.
Parameters
==========
graph_id : GraphID
Graph ID of the desired component.
"""
return self._components_by_id.get(graph_id, None)
......@@ -524,8 +533,11 @@ class SFG(AbstractOperation):
"""Find all graph components with the specified name.
Returns an empty sequence if no components were found.
Keyword arguments:
name: Name of the desired component(s)
Parameters
==========
name : Name
Name of the desired component(s)
"""
return self._components_by_name.get(name, [])
......@@ -536,9 +548,13 @@ class SFG(AbstractOperation):
return a sequence of the keys to use when fetching their results
from a simulation.
Keyword arguments:
name: Name of the desired component(s)
output_index: The desired output index to get the result from
Parameters
==========
name : Name
Name of the desired component(s)
output_index : int, default: 0
The desired output index to get the result from
"""
keys = []
for comp in self.find_by_name(name):
......@@ -553,9 +569,11 @@ class SFG(AbstractOperation):
Find and replace all components matching either on GraphID, Type or both.
Then return a new deepcopy of the sfg with the replaced component.
Arguments:
component: The new component(s), e.g. Multiplication
graph_id: The GraphID to match the component to replace.
Parameters
==========
component : The new component(s), e.g. Multiplication
graph_id : The GraphID to match the component to replace.
"""
sfg_copy = self() # Copy to not mess with this SFG.
......@@ -586,13 +604,17 @@ class SFG(AbstractOperation):
def insert_operation(
self, component: Operation, output_comp_id: GraphID
) -> Optional["SFG"]:
"""Insert an operation in the SFG after a given source operation.
The source operation output count must match the input count of the operation as well as the output
"""
Insert an operation in the SFG after a given source operation.
The source operation output count must match the input count of the operation
as well as the output.
Then return a new deepcopy of the sfg with the inserted component.
Arguments:
component: The new component, e.g. Multiplication.
output_comp_id: The source operation GraphID to connect from.
Parameters
==========
component : The new component, e.g. Multiplication.
output_comp_id : The source operation GraphID to connect from.
"""
# Preserve the original SFG by creating a copy.
......@@ -621,7 +643,7 @@ class SFG(AbstractOperation):
# Recreate the newly coupled SFG so that all attributes are correct.
return sfg_copy()
def remove_operation(self, operation_id: GraphID) -> "SFG":
def remove_operation(self, operation_id: GraphID) -> Union["SFG", None]:
"""Returns a version of the SFG where the operation with the specified GraphID removed.
The operation has to have the same amount of input- and output ports or a ValueError will
be raised. If no operation with the entered operation_id is found then returns None and does nothing.
......@@ -955,7 +977,6 @@ class SFG(AbstractOperation):
while op_stack:
original_op = op_stack.pop()
# Add or get the new copy of the operation.
new_op = None
if original_op not in self._original_components_to_new:
new_op = self._add_component_unconnected_copy(original_op)
self._components_dfs_order.append(new_op)
......
"""B-ASIC Simulation Module.
"""
B-ASIC Simulation Module.
Contains a class for simulating the result of an SFG given a set of input values.
"""
......@@ -35,7 +36,8 @@ InputProvider = Union[Number, Sequence[Number], InputFunction]
class Simulation:
"""Simulation of an SFG.
"""
Simulation of an SFG.
Use FastSimulation (from the C++ extension module) for a more effective
simulation when running many iterations.
......@@ -68,7 +70,9 @@ class Simulation:
self.set_inputs(input_providers)
def set_input(self, index: int, input_provider: InputProvider) -> None:
"""Set the input function used to get values for the specific input at the given index to the internal SFG.
"""
Set the input function used to get values for the specific input at the
given index to the internal SFG.
"""
if index < 0 or index >= len(self._input_functions):
raise IndexError(
......@@ -92,7 +96,8 @@ class Simulation:
def set_inputs(
self, input_providers: Sequence[Optional[InputProvider]]
) -> None:
"""Set the input functions used to get values for the inputs to the internal SFG.
"""
Set the input functions used to get values for the inputs to the internal SFG.
"""
if len(input_providers) != self._sfg.input_count:
raise ValueError(
......@@ -109,7 +114,8 @@ class Simulation:
bits_override: Optional[int] = None,
truncate: bool = True,
) -> Sequence[Number]:
"""Run one iteration of the simulation and return the resulting output values.
"""
Run one iteration of the simulation and return the resulting output values.
"""
return self.run_for(1, save_results, bits_override, truncate)
......@@ -120,7 +126,8 @@ class Simulation:
bits_override: Optional[int] = None,
truncate: bool = True,
) -> Sequence[Number]:
"""Run the simulation until its iteration is greater than or equal to the given iteration
"""
Run the simulation until its iteration is greater than or equal to the given iteration
and return the output values of the last iteration.
"""
result = []
......@@ -151,7 +158,9 @@ class Simulation:
bits_override: Optional[int] = None,
truncate: bool = True,
) -> Sequence[Number]:
"""Run a given number of iterations of the simulation and return the output values of the last iteration.
"""
Run a given number of iterations of the simulation and return the output
values of the last iteration.
"""
return self.run_until(
self._iteration + iterations, save_results, bits_override, truncate
......@@ -163,7 +172,9 @@ class Simulation:
bits_override: Optional[int] = None,
truncate: bool = True,
) -> Sequence[Number]:
"""Run the simulation until the end of its input arrays and return the output values of the last iteration.
"""
Run the simulation until the end of its input arrays and return the output
values of the last iteration.
"""
if self._input_length is None:
raise IndexError("Tried to run unlimited simulation")
......@@ -178,10 +189,14 @@ class Simulation:
@property
def results(self) -> ResultArrayMap:
"""Get a mapping from result keys to numpy arrays containing all results, including intermediate values,
calculated for each iteration up until now that was run with save_results enabled.
"""
Get a mapping from result keys to numpy arrays containing all results, including
intermediate values, calculated for each iteration up until now that was run with
save_results enabled.
The mapping is indexed using the key() method of Operation with the appropriate output index.
Example result after 3 iterations: {"c1": [3, 6, 7], "c2": [4, 5, 5], "bfly1.0": [7, 0, 0], "bfly1.1": [-1, 0, 2], "0": [7, -2, -1]}
Example result after 3 iterations::
{"c1": [3, 6, 7], "c2": [4, 5, 5], "bfly1.0": [7, 0, 0], "bfly1.1": [-1, 0, 2], "0": [7, -2, -1]}
"""
return {key: np.array(value) for key, value in self._results.items()}
......@@ -190,6 +205,7 @@ class Simulation:
self._results.clear()
def clear_state(self) -> None:
"""Clear all current state of the simulation, except for the results and iteration.
"""
Clear all current state of the simulation, except for the results and iteration.
"""
self._delays.clear()
"""B-ASIC Special Operations Module.
"""
B-ASIC Special Operations Module.
Contains operations with special purposes that may be treated differently from
normal operations in an SFG.
......@@ -19,7 +20,8 @@ from b_asic.port import SignalSourceProvider
class Input(AbstractOperation):
"""Input operation.
"""
Input operation.
Marks an input port to an SFG.
Its value will be updated on each iteration when simulating the SFG.
......@@ -83,7 +85,8 @@ class Input(AbstractOperation):
class Output(AbstractOperation):
"""Output operation.
"""
Output operation.
Marks an output port to an SFG.
The SFG will forward its input to the corresponding output signal
......@@ -126,7 +129,8 @@ class Output(AbstractOperation):
class Delay(AbstractOperation):
"""Unit delay operation.
"""
Unit delay operation.
Represents one unit of delay in a circuit, typically a clock cycle.
Can be thought of as a register or a D flip-flop.
......
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