diff --git a/b_asic/architecture.py b/b_asic/architecture.py
index 2a9a6074cafff63261b5ab6da1c522c3f6083511..d675dded72de660e28c2dd445229d6d00e50b2a9 100644
--- a/b_asic/architecture.py
+++ b/b_asic/architecture.py
@@ -3,7 +3,18 @@ B-ASIC architecture classes.
"""
from collections import defaultdict
from io import TextIOWrapper
-from typing import Dict, Iterable, Iterator, List, Optional, Set, Tuple, Union, cast
+from typing import (
+ DefaultDict,
+ Dict,
+ Iterable,
+ Iterator,
+ List,
+ Optional,
+ Set,
+ Tuple,
+ Union,
+ cast,
+)
import matplotlib.pyplot as plt
from graphviz import Digraph
@@ -30,11 +41,11 @@ class HardwareBlock:
"""
def __init__(self, entity_name: Optional[str] = None):
- self._entity_name = None
+ self._entity_name: Optional[str] = None
if entity_name is not None:
self.set_entity_name(entity_name)
- def set_entity_name(self, entity_name: str):
+ def set_entity_name(self, entity_name: str) -> None:
"""
Set entity name of hardware block.
@@ -44,7 +55,7 @@ class HardwareBlock:
The entity name.
"""
if not is_valid_vhdl_identifier(entity_name):
- raise ValueError(f'{entity_name} is not a valid VHDL indentifier')
+ raise ValueError(f'{entity_name} is not a valid VHDL identifier')
self._entity_name = entity_name
def write_code(self, path: str) -> None:
@@ -56,7 +67,7 @@ class HardwareBlock:
path : str
Directory to write code in.
"""
- if not self.entity_name:
+ if not self._entity_name:
raise ValueError("Entity name must be set")
raise NotImplementedError
@@ -173,16 +184,41 @@ class Resource(HardwareBlock):
# doc-string inherited
return self._collection.schedule_time
- def plot_content(self, ax: plt.Axes) -> None:
+ def plot_content(self, ax: plt.Axes, **kwargs) -> None:
+ """
+ Plot the content of the resource.
+
+ This plots the assigned processes executed on this resource.
+
+ Parameters
+ ----------
+ ax : Axes
+ Matplotlib Axes to plot in.
+ **kwargs
+ Passed to :meth:`b_asic.resources.ProcessCollection.plot`
+ """
if not self.is_assigned:
- self._collection.plot(ax)
+ self._collection.plot(ax, **kwargs)
else:
for i, pc in enumerate(self._assignment): # type: ignore
- pc.plot(ax=ax, row=i)
+ pc.plot(ax=ax, row=i, **kwargs)
- def show_content(self):
+ def show_content(self, title=None, **kwargs) -> None:
+ """
+ Display the content of the resource.
+
+ This displays the assigned processes executed on this resource.
+
+ Parameters
+ ----------
+ title : str, optional
+ **kwargs
+ Passed to :meth:`b_asic.resources.ProcessCollection.plot`
+ """
fig, ax = plt.subplots()
- self.plot_content(ax)
+ self.plot_content(ax, **kwargs)
+ if title:
+ fig.suptitle(title)
fig.show() # type: ignore
@property
@@ -201,6 +237,10 @@ class Resource(HardwareBlock):
self.plot_content(ax)
return fig
+ @property
+ def collection(self) -> ProcessCollection:
+ return self._collection
+
class ProcessingElement(Resource):
"""
@@ -233,10 +273,8 @@ class ProcessingElement(Resource):
op_type = type(ops[0])
if not all(isinstance(op, op_type) for op in ops):
raise TypeError("Different Operation types in ProcessCollection")
- self._collection = process_collection
self._operation_type = op_type
self._type_name = op_type.type_name()
- self._entity_name = entity_name
self._input_count = ops[0].input_count
self._output_count = ops[0].output_count
self._assignment = list(
@@ -246,8 +284,8 @@ class ProcessingElement(Resource):
raise ValueError("Cannot map ProcessCollection to single ProcessingElement")
@property
- def processes(self) -> Set[OperatorProcess]:
- return {cast(OperatorProcess, p) for p in self._collection}
+ def processes(self) -> List[OperatorProcess]:
+ return [cast(OperatorProcess, p) for p in self._collection]
class Memory(Resource):
@@ -322,6 +360,27 @@ class Memory(Resource):
self._collection.split_on_execution_time(heuristic=heuristic)
)
+ def assign(self, heuristic: str = "left_edge") -> None:
+ """
+ Perform assignment of the memory variables.
+
+ Parameters
+ ----------
+ heuristic : str
+ The assignment algorithm. Depending on memory type the following are
+ available:
+
+ * 'RAM'
+ * 'left_edge': Left-edge algorithm.
+ * 'graph_color': Graph-coloring based on exclusion graph.
+ * 'register'
+ * ...
+ """
+ if self._memory_type == "RAM":
+ self._assign_ram(heuristic=heuristic)
+ else: # "register"
+ raise NotImplementedError()
+
class Architecture(HardwareBlock):
"""
@@ -350,11 +409,11 @@ of :class:`~b_asic.architecture.ProcessingElement`
):
super().__init__(entity_name)
self._processing_elements = (
- set(processing_elements)
+ [processing_elements]
if isinstance(processing_elements, ProcessingElement)
- else processing_elements
+ else list(processing_elements)
)
- self._memories = set(memories) if isinstance(memories, Memory) else memories
+ self._memories = [memories] if isinstance(memories, Memory) else list(memories)
self._direct_interconnects = direct_interconnects
self._variable_inport_to_resource: Dict[InputPort, Tuple[Resource, int]] = {}
self._variable_outport_to_resource: Dict[OutputPort, Tuple[Resource, int]] = {}
@@ -380,7 +439,7 @@ of :class:`~b_asic.architecture.ProcessingElement`
raise ValueError(f"Different schedule times: {schedule_times}")
return schedule_times.pop()
- def _build_dicts(self):
+ def _build_dicts(self) -> None:
for pe in self.processing_elements:
for operator in pe.processes:
for input_port in operator.operation.inputs:
@@ -406,7 +465,7 @@ of :class:`~b_asic.architecture.ProcessingElement`
read_port.index,
)
- def validate_ports(self):
+ def validate_ports(self) -> None:
# Validate inputs and outputs of memory variables in all the memories in this
# architecture
memory_read_ports = set()
@@ -442,7 +501,9 @@ of :class:`~b_asic.architecture.ProcessingElement`
)
# Make sure all inputs and outputs in the architecture are in use
- def get_interconnects_for_memory(self, mem: Memory):
+ def get_interconnects_for_memory(
+ self, mem: Memory
+ ) -> Tuple[Dict[Resource, int], Dict[Resource, int]]:
"""
Return a dictionary with interconnect information for a Memory.
@@ -457,9 +518,9 @@ of :class:`~b_asic.architecture.ProcessingElement`
A dictionary with the ProcessingElements that are connected to the write and
read ports, respectively, with counts of the number of accesses.
"""
- d_in = defaultdict(_interconnect_dict)
- d_out = defaultdict(_interconnect_dict)
- for var in mem._collection:
+ d_in: DefaultDict[Resource, int] = defaultdict(_interconnect_dict)
+ d_out: DefaultDict[Resource, int] = defaultdict(_interconnect_dict)
+ for var in mem.collection:
var = cast(MemoryVariable, var)
d_in[self._operation_outport_to_resource[var.write_port]] += 1
for read_port in var.read_ports:
@@ -490,21 +551,23 @@ of :class:`~b_asic.architecture.ProcessingElement`
frequency of accesses.
"""
- ops = cast(List[OperatorProcess], list(pe._collection))
- d_in = [defaultdict(_interconnect_dict) for _ in ops[0].operation.inputs]
- d_out = [defaultdict(_interconnect_dict) for _ in ops[0].operation.outputs]
- for var in pe._collection:
+ d_in: List[DefaultDict[Tuple[Resource, int], int]] = [
+ defaultdict(_interconnect_dict) for _ in range(pe.input_count)
+ ]
+ d_out: List[DefaultDict[Tuple[Resource, int], int]] = [
+ defaultdict(_interconnect_dict) for _ in range(pe.output_count)
+ ]
+ for var in pe.collection:
var = cast(OperatorProcess, var)
- for i, input in enumerate(var.operation.inputs):
- d_in[i][self._variable_inport_to_resource[input]] += 1
+ for i, input_ in enumerate(var.operation.inputs):
+ d_in[i][self._variable_inport_to_resource[input_]] += 1
for i, output in enumerate(var.operation.outputs):
d_out[i][self._variable_outport_to_resource[output]] += 1
return [dict(d) for d in d_in], [dict(d) for d in d_out]
def _digraph(self) -> Digraph:
- edges = set()
+ edges: Set[Tuple[str, str, str]] = set()
dg = Digraph(node_attr={'shape': 'record'})
- # dg.attr(rankdir="LR")
for i, mem in enumerate(self._memories):
dg.node(mem.entity_name, mem._struct_def())
for i, pe in enumerate(self._processing_elements):
@@ -520,8 +583,8 @@ of :class:`~b_asic.architecture.ProcessingElement`
f"{cnt}",
)
)
- for o, outp in enumerate(outputs):
- for (dest, port), cnt in outp.items():
+ for o, output in enumerate(outputs):
+ for (dest, port), cnt in output.items():
edges.add(
(
f"{pe.entity_name}:out{o}",
@@ -529,16 +592,16 @@ of :class:`~b_asic.architecture.ProcessingElement`
f"{cnt}",
)
)
- for src, dest, cnt in edges:
- dg.edge(src, dest, label=cnt)
+ for src_str, dest_str, cnt_str in edges:
+ dg.edge(src_str, dest_str, label=cnt_str)
return dg
@property
- def memories(self) -> Iterable[Memory]:
+ def memories(self) -> List[Memory]:
return self._memories
@property
- def processing_elements(self) -> Iterable[ProcessingElement]:
+ def processing_elements(self) -> List[ProcessingElement]:
return self._processing_elements
@property
diff --git a/docs_sphinx/conf.py b/docs_sphinx/conf.py
index f8148bc4a1b069a0368a746ed085dce0147a851f..7f335128f716a8eb2c9ad8ff62e68b698f909c50 100644
--- a/docs_sphinx/conf.py
+++ b/docs_sphinx/conf.py
@@ -8,7 +8,7 @@
import shutil
-import qtgallery
+# import qtgallery
project = 'B-ASIC'
copyright = '2020-2023, Oscar Gustafsson et al'
@@ -28,7 +28,7 @@ extensions = [
'sphinx_gallery.gen_gallery',
'numpydoc', # Needs to be loaded *after* autodoc.
'jupyter_sphinx',
- 'qtgallery',
+ # 'qtgallery',
]
templates_path = ['_templates']
@@ -71,11 +71,11 @@ sphinx_gallery_conf = {
'doc_module': ('b_asic',),
'reference_url': {'b_asic': None},
'image_scrapers': (
- qtgallery.qtscraper,
+ # qtgallery.qtscraper,
'matplotlib',
),
'reset_modules': (
- qtgallery.reset_qapp,
+ # qtgallery.reset_qapp,
'matplotlib',
),
}
diff --git a/examples/secondorderdirectformiir_architecture.py b/examples/secondorderdirectformiir_architecture.py
index f4e240a4d1241c225e346d5b02c0351a966a4a68..82b50d2f45155e762cfa434981ec51b389579f1f 100644
--- a/examples/secondorderdirectformiir_architecture.py
+++ b/examples/secondorderdirectformiir_architecture.py
@@ -42,7 +42,7 @@ schedule = Schedule(sfg, cyclic=True)
schedule.show(title='Original schedule')
# %%
-# Rescheudle to only require one adder and one multiplier
+# Reschedule to only require one adder and one multiplier
schedule.move_operation('add4', 2)
schedule.move_operation('cmul5', -4)
schedule.move_operation('cmul4', -5)
@@ -68,18 +68,22 @@ p_in = ProcessingElement(inputs, entity_name='input')
p_out = ProcessingElement(outputs, entity_name='output')
# %%
-# Extract memory variables
+# Extract and assign memory variables
mem_vars = schedule.get_memory_variables()
mem_vars.show(title="All memory variables")
direct, mem_vars = mem_vars.split_on_length()
-direct.show(title="Direct interconnects")
mem_vars.show(title="Non-zero time memory variables")
mem_vars_set = mem_vars.split_on_ports(read_ports=1, write_ports=1, total_ports=2)
-memories = set()
+memories = []
for i, mem in enumerate(mem_vars_set):
- memories.add(Memory(mem, entity_name=f"memory{i}"))
- mem.show(title=f"memory{i}")
+ memory = Memory(mem, memory_type="RAM", entity_name=f"memory{i}")
+ memories.append(memory)
+ mem.show(title=f"{memory.entity_name}")
+ memory.assign("left_edge")
+ memory.show_content(title=f"Assigned {memory.entity_name}")
+
+direct.show(title="Direct interconnects")
# %%
# Create architecture