`b_asic.resources`¶

B-ASIC Resources Module.

Contains functionality for grouping processes into collections.

class b_asic.resources.ProcessCollection(collection: Iterable[Process], schedule_time: int, cyclic: bool = False)¶

Collection of Process objects.

Parameters:

collectionIterable of Process objects: The Process objects forming this ProcessCollection.
schedule_timeint: The scheduling time associated with this ProcessCollection.
cyclicbool, default: False: Whether the processes operate cyclically, i.e., if time

\[t = t \bmod T_{\textrm{schedule}}.\]

add_process(process: Process) → None¶

Add a Process.

Parameters:

processProcess: The Process object to add.

property collection: list[Process]¶

exclusion_graph_from_execution_time() → Graph¶

Create an exclusion graph from processes overlapping in execution time.

Returns:

networkx.Graph

exclusion_graph_from_ports(read_ports: int | None = None, write_ports: int | None = None, total_ports: int | None = None) → Graph¶

Create an exclusion graph based on concurrent read and write accesses.

Parameters:

read_portsint: The number of read ports used when splitting process collection based on memory variable access.
write_portsint: The number of write ports used when splitting process collection based on memory variable access.
total_portsint: The total number of ports used when splitting process collection based on memory variable access.

Returns:

networkx.Graph: An undirected exclusion graph.

find_by_time(time: int) → ProcessCollection¶

from_name(name: str) → Process¶

Get a Process from its name.

Parameters:

namestr: The name of the process to retrieve.

Raises:

KeyError: If no processes with name is found in this collection.

generate_memory_based_storage_vhdl(filename: str, entity_name: str, word_length: int, assignment: list[ProcessCollection], read_ports: int = 1, write_ports: int = 1, total_ports: int = 2, *, input_sync: bool = True, adr_mux_size: int | None = None, adr_pipe_depth: int | None = None) → None¶

Generate VHDL code for memory-based storage of processes (MemoryVariables).

Parameters:

filenamestr: Filename of output file.
entity_namestr: Name used for the VHDL entity.
word_lengthint: Word length of the memory variable objects.
assignmentlist: A possible cell assignment to use when generating the memory based storage. The cell assignment is a dictionary int to ProcessCollection where the integer corresponds to the cell to assign all MemoryVariables in corresponding process collection. If unset, each MemoryVariable will be assigned to a unique single cell.
read_portsint, default: 1: The number of read ports used when splitting process collection based on memory variable access. If total ports in unset, this parameter has to be set and total_ports is assumed to be read_ports + write_ports.
write_portsint, default: 1: The number of write ports used when splitting process collection based on memory variable access. If total ports is unset, this parameter has to be set and total_ports is assumed to be read_ports + write_ports.
total_portsint, default: 2: The total number of ports used when splitting process collection based on memory variable access.
input_syncbool, default: True: Add registers to the input signals (enable signal and data input signals). Adding registers to the inputs allow pipelining of address generation (which is added automatically). For large interleavers, this can improve timing significantly.
adr_mux_sizeint, optional: Size of multiplexer if using address generation pipelining. Set to None for no multiplexer pipelining. If any other value than None, input_sync must also be set.
adr_pipe_depthint, optional: Depth of address generation pipelining. Set to None for no multiplexer pipelining. If any other value than None, input_sync must also be set.

generate_register_based_storage_vhdl(filename: str, word_length: int, entity_name: str, read_ports: int = 1, write_ports: int = 1, total_ports: int = 2) → None¶

Generate VHDL code for register-based storage of processes (MemoryVariables).

This is based on Forward-Backward Register Allocation.

Parameters:

filenamestr: Filename of output file.
word_lengthint: Word length of the memory variable objects.
entity_namestr: Name used for the VHDL entity.
read_portsint, default: 1: The number of read ports used when splitting process collection based on memory variable access. If total ports in unset, this parameter has to be set and total_ports is assumed to be read_ports + write_ports.
write_portsint, default: 1: The number of write ports used when splitting process collection based on memory variable access. If total ports is unset, this parameter has to be set and total_ports is assumed to be read_ports + write_ports.
total_portsint, default: 2: The total number of ports used when splitting process collection based on memory variable access.

References

K. Parhi: VLSI Digital Signal Processing Systems: Design and Implementation, Ch. 6.3.2

get_by_type(process_type: type[Process]) → ProcessCollection¶

Return a new ProcessCollection containing only processes of the given type.

Parameters:

process_typetype: The Process subclass to filter by.

Returns:

A new ProcessCollection.

get_by_type_name(type_name: TypeName) → ProcessCollection¶

Return a new ProcessCollection containing only processes of the given type name.

Parameters:

type_nameTypeName: The TypeName of the operation to extract.

Returns:

A new ProcessCollection.

plot(ax: Axes | None = None, *, show_name: bool = True, bar_color: str | tuple[float, ...] = (0.0, 0.7254901960784313, 0.9058823529411765), marker_color: str | tuple[float, ...] = 'black', marker_read: str = 'X', marker_write: str = 'o', show_markers: bool = True, row: int | None = None, allow_excessive_lifetimes: bool = False) → Axes¶

Plot lifetime diagram.

Plot all Process objects of this ProcessCollection in a lifetime diagram.

If the ax parameter is not specified, a new Matplotlib figure is created.

Raises KeyError if any Process lifetime exceeds this ProcessCollection schedule time, unless allow_excessive_lifetimes is True. In that case, Process objects whose lifetime exceed the schedule time are drawn using the B-ASIC warning color.

Parameters:

axmatplotlib.axes.Axes, optional: Matplotlib Axes object to draw this lifetime chart onto. If not provided (i.e., set to None), this method will return a new Axes object.
show_namebool, default: True: Show name of all processes in the lifetime chart.
bar_colorcolor, optional: Bar color in lifetime chart.
marker_colorcolor, default ‘black’: Color for read and write marker.
marker_readstr, default ‘o’: Marker at read time in the lifetime chart.
marker_writestr, default ‘x’: Marker at write time in the lifetime chart.
show_markersbool, default True: Show markers at read and write times.
rowint, optional: Render all processes in this collection on a specified row in the Matplotlib axes object. Defaults to None, which renders all processes on separate rows. This option is useful when drawing cell assignments.
allow_excessive_lifetimesbool, default False: If set to true, the plot method allows plotting collections of variables with a longer lifetime than the schedule time.

Returns:

axmatplotlib.axes.Axes: Associated Matplotlib Axes (or array of Axes) object.

plot_port_accesses(axes) → None¶

Plot read, write, and total accesses.

These are plotted as bar graphs.

Parameters:

axeslist of three matplotlib.axes.Axes: Three Axes to plot in.

plot_total_execution_times(ax) → None¶

Plot total execution times for each time slot.

This is plotted as a bar graph.

Parameters:

axmatplotlib.axes.Axes: The Axes to plot in.

processing_element_bound() → int¶

Get the lower-bound on the number of processing elements.

Returns:

int: The maximum number of concurrent executions times.

read_port_accesses() → dict[int, int]¶

read_ports_bound() → int¶

Get the lower-bound on the number of read ports.

Returns:

int: The maximum number of concurrent reads.

remove_process(process: Process) → None¶

Remove a Process.

Raises KeyError if the specified Process is not in this collection.

Parameters:

processProcess: The Process object to remove from this collection.

property schedule_time: int¶

show(*, show_name: bool = True, bar_color: str | tuple[float, ...] = (0.0, 0.7254901960784313, 0.9058823529411765), marker_color: str | tuple[float, ...] = 'black', marker_read: str = 'X', marker_write: str = 'o', show_markers: bool = True, allow_excessive_lifetimes: bool = False, title: str | None = None) → None¶

Display lifetime diagram using the current Matplotlib backend.

Equivalent to creating a Matplotlib figure, passing it and arguments to plot() and invoking matplotlib.figure.Figure.show().

Parameters:

show_namebool, default: True: Show name of all processes in the lifetime chart.
bar_colorcolor, optional: Bar color in lifetime chart.
marker_colorcolor, default ‘black’: Color for read and write marker.
marker_readstr, default ‘o’: Marker at read time in the lifetime chart.
marker_writestr, default ‘x’: Marker at write time in the lifetime chart.
show_markersbool, default True: Show markers at read and write times.
allow_excessive_lifetimesbool, default False: If True, the plot method allows plotting collections of variables with a greater lifetime than the schedule time.
titlestr, optional: Figure title.

show_port_accesses(title: str = '') → None¶

Show read, write, and total accesses.

Parameters:

titlestr, optional: Figure title.

show_total_execution_times(title: str = '') → None¶

Show total execution time for each time slot.

Parameters:

titlestr, optional: Figure title.

split_into_chains(max_length: int | None = None) → ProcessCollection¶

Split long-lived memory variables into chains of shorter variables.

Each MemoryVariable with execution_time > max_length is replaced by a chain of variables, each with execution_time <= max_length.

Parameters:

max_lengthint, optional: Maximum allowed execution time for any variable in the result. Defaults to self.schedule_time.

Returns:

ProcessCollection: New collection with the same schedule_time and cyclic settings, but all variables within max_length.

split_on_execution_time(strategy: Literal['left_edge', 'left_edge_min_mux', 'greedy_graph_color', 'ilp_graph_color'] = 'left_edge', alg_params: dict | None = None) → list[ProcessCollection]¶

Split based on overlapping execution time.

Parameters:

strategy{‘ilp_graph_color’, ‘greedy_graph_color’, ‘left_edge’, ‘left_edge_min_mux’}, default: ‘left_edge’

The strategy used when splitting based on execution times.

alg_paramsdict, optional

Algorithm-specific parameters. Valid keys depend on strategy:

For 'greedy_graph_color':

coloring_strategystr, default: 'saturation_largest_first': Node ordering strategy passed to networkx.algorithms.coloring.greedy_color(). Valid values are 'largest_first', 'random_sequential', 'smallest_last', 'independent_set', 'connected_sequential_bfs', 'connected_sequential_dfs', 'connected_sequential', 'saturation_largest_first', 'DSATUR'.

For 'ilp_graph_color':

max_colorsint, optional

The maximum number of colors (resources) to split into.

solverLpSolver, optional

ILP solver to use. To see available solvers:

import pulp

print(pulp.listSolvers(onlyAvailable=True))

For 'left_edge_min_mux':

direct_variablesProcessCollection: Variables that are passed directly between operations without going through memory. Used to inform mux-minimizing placement decisions.

Returns:

A list of new ProcessCollection objects with the process splitting.

split_on_length(length: int = 0) → tuple[ProcessCollection, ProcessCollection]¶

Split into two ProcessCollections based on execution time length.

Parameters:

lengthint, default: 0: The execution time length to split on. Length is inclusive for the smaller collection.

Returns:

tuple(ProcessCollection, ProcessCollection): A tuple of two ProcessCollections, one with shorter than or equal execution times and one with longer execution times.

split_on_ports(strategy: Literal['ilp_graph_color', 'ilp_min_input_mux', 'ilp_min_output_mux', 'ilp_min_mux', 'greedy_graph_color', 'equitable_graph_color', 'left_edge', 'left_edge_min_pe_to_mem', 'left_edge_min_mem_to_pe', 'left_edge_min_mux'] = 'left_edge', read_ports: int | None = None, write_ports: int | None = None, total_ports: int | None = None, alg_params: dict | None = None) → list[ProcessCollection]¶

Split based on concurrent read and write accesses.

Different strategy methods can be used.

Parameters:

strategystr, default: 'left_edge'

The strategy used when splitting this ProcessCollection. Valid options are:

'ilp_graph_color' - ILP-based optimal graph coloring.
'ilp_min_input_mux' - ILP-based optimal graph coloring reducing the number of PE -> memory multiplexers.
'ilp_min_output_mux' - ILP-based optimal graph coloring reducing the number of memory -> PE multiplexers.
'ilp_min_mux' - ILP-based optimal graph coloring reducing the number of total multiplexers.
'greedy_graph_color' - Greedy graph coloring based heuristic.
'equitable_graph_color' - Equitable graph coloring, attempting to divide the variables evenly.
'left_edge' - Greedy heuristic for assigning variables.
'left_edge_min_pe_to_mem' - Greedy heuristic for assigning variables, attempting to reduce the amount of PE -> memory connections.
'left_edge_min_mem_to_pe' - Greedy heuristic for assigning variables, attempting to reduce the amount of memory -> PE connections.
'left_edge_min_mux' - Greedy heuristic for assigning variables, attempting to reduce the total number of multiplexers.

read_portsint, optional

The number of read ports per memory resource.

write_portsint, optional

The number of write ports per memory resource.

total_portsint, optional

The total number of ports per memory resource.

alg_paramsdict, optional

Algorithm-specific parameters. Valid keys depend on strategy:

For 'ilp_graph_color', 'ilp_min_input_mux', 'ilp_min_output_mux', 'ilp_min_mux':

max_colorsint, optional

The maximum number of colors (memory resources) to split into.

solverLpSolver, optional

ILP solver to use. To see available solvers:

import pulp

print(pulp.listSolvers(onlyAvailable=True))

For 'ilp_min_input_mux', 'ilp_min_output_mux', 'ilp_min_mux', 'left_edge_min_pe_to_mem', 'left_edge_min_mem_to_pe':

processing_elementslist of ProcessingElement: The currently used processing elements. Used to determine PE-memory connections when minimizing multiplexers.

For 'left_edge_min_mux':

direct_variablesProcessCollection: Variables that are passed directly between operations without going through memory. Used to inform mux-minimizing placement decisions.

Returns:

A list of new ProcessCollection objects with the process splitting.

split_on_type_name() → dict[TypeName, ProcessCollection]¶

split_ports_sequentially(read_ports: int, write_ports: int, total_ports: int, sequence: list[Process]) → list[ProcessCollection]¶

Split this collection by sequentially assigning processes in the order of sequence.

This method takes the processes from sequence, in order, and assigns them to to multiple new ProcessCollection based on port collisions in a first-come first-served manner. The first Process in sequence is assigned first, and the last Process in sequence is assigned last.

Parameters:

read_portsint: The number of read ports used when splitting process collection based on memory variable access.
write_portsint: The number of write ports used when splitting process collection based on memory variable access.
total_portsint: The total number of ports used when splitting process collection based on memory variable access.
sequencelist of Process: A list of the processes used to determine the order in which processes are assigned.

Returns:

list of ProcessCollection: A list of new ProcessCollection objects with the process splitting.

total_execution_times() → dict[int, int]¶

total_port_accesses() → dict[int, int]¶

total_ports_bound() → int¶

Get the lower-bound on the number of total ports.

Returns:

int: The maximum number of concurrent reads and writes.

write_port_accesses() → dict[int, int]¶

write_ports_bound() → int¶

Get the lower-bound on the number of write ports.

Returns:

int: The maximum number of concurrent writes.

b_asic.resources.draw_exclusion_graph_coloring(exclusion_graph: Graph, color_dict: dict[Process, int], ax: Axes | None = None, color_list: list[str] | list[tuple[float, float, float]] | None = None, **kwargs) → None¶

Draw the colored exclusion graphs.

Example usage:

import networkx as nx
import matplotlib.pyplot as plt

fig, ax = plt.subplots()
collection = ProcessCollection(...)
exclusion_graph = collection.exclusion_graph_from_ports(
    read_ports=1,
    write_ports=1,
    total_ports=2,
)
coloring = nx.greedy_color(exclusion_graph)
draw_exclusion_graph_coloring(exclusion_graph, coloring, ax=ax)
fig.show()

Parameters:

exclusion_graphnetworkx.Graph: The networkx.Graph exclusion graph object that is to be drawn.
color_dictdict: A dict where keys are Process objects and values are integers representing colors. These dictionaries are automatically generated by networkx.algorithms.coloring.greedy_color().
axmatplotlib.axes.Axes, optional: A Matplotlib Axes object to draw the exclusion graph.
color_listiterable of color, optional: A list of colors in Matplotlib format.
**kwargsAny: Named arguments passed on to networkx.draw_networkx()

b_asic.resources¶

`b_asic.resources`¶