`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.

get_by_type_name(type_name: TypeName) → ProcessCollection¶

Return a new ProcessCollection with only a given type of operation.

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_on_execution_time(strategy: Literal['left_edge', 'greedy_graph_color', 'ilp_graph_color'] = 'left_edge', coloring_strategy: Literal['largest_first', 'random_sequential', 'smallest_last', 'independent_set', 'connected_sequential_bfs', 'connected_sequential_dfs', 'connected_sequential', 'saturation_largest_first', 'DSATUR'] = 'saturation_largest_first', max_colors: int | None = None, solver: LpSolver | None = None) → list[ProcessCollection]¶

Split based on overlapping execution time.

Parameters:

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

The strategy used when splitting based on execution times.

coloring_strategystr, default: ‘saturation_largest_first’

Node ordering strategy passed to networkx.algorithms.coloring.greedy_color(). This parameter is only considered if strategy is set to ‘greedy_graph_color’. Valid options are:

‘largest_first’
‘random_sequential’
‘smallest_last’
‘independent_set’
‘connected_sequential_bfs’
‘connected_sequential_dfs’ or ‘connected_sequential’
‘saturation_largest_first’ or ‘DSATUR’

max_colorsint, optional

The maximum amount of colors to split based on, only required if strategy is an ILP method.

solverLpSolver, optional

Only used if strategy is an ILP method. To see which solvers are available:

import pulp

print(pulp.listSolvers(onlyAvailable=True))

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_total_mux', 'greedy_graph_color', 'equitable_graph_color', 'left_edge', 'left_edge_min_pe_to_mem', 'left_edge_min_mem_to_pe'] = 'left_edge', read_ports: int | None = None, write_ports: int | None = None, total_ports: int | None = None, processing_elements: list[ProcessingElement] | None = None, max_colors: int | None = None, solver: LpSolver | 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_total_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.

read_portsint, optional

The number of read ports used when splitting process collection based on memory variable access.

write_portsint, optional

The number of write ports used when splitting process collection based on memory variable access.

total_portsint, optional

The total number of ports used when splitting process collection based on memory variable access.

processing_elementslist of ProcessingElement, optional

The currently used PEs, only required if strategy is “left_edge_min_mem_to_pe”, “ilp_graph_color” or “ilp_min_input_mux”.

max_colorsint, optional

The maximum amount of colors to split based on, only required if strategy is an ILP method.

solverLpSolver, optional

Only used if strategy is an ILP method. To see which solvers are available:

import pulp

print(pulp.listSolvers(onlyAvailable=True))

Returns:

A set 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`¶