architecture.py: add assignments of Resources

b_asic/architecture.py

@@ -5,6 +5,7 @@ from collections import defaultdict
 from io import TextIOWrapper
 from typing import Dict, Iterable, Iterator, List, Optional, Set, Tuple, Union, cast
 
+import matplotlib.pyplot as plt
 from graphviz import Digraph
 
 from b_asic.port import InputPort, OutputPort
@@ -132,6 +133,7 @@ class Resource(HardwareBlock):
         self._collection = process_collection
         self._input_count = -1
         self._output_count = -1
+        self._assignment: Optional[List[ProcessCollection]] = None
 
     def __repr__(self):
         return self.entity_name
@@ -172,6 +174,19 @@ class Resource(HardwareBlock):
         # doc-string inherited
         return self._collection.schedule_time
 
+    def show_content(self):
+        if not self.is_assigned:
+            self._collection.show()
+        else:
+            fig, ax = plt.subplots()
+            for i, pc in enumerate(self._assignment):  # type: ignore
+                pc.plot(ax=ax, row=i)
+            fig.show()  # type: ignore
+
+    @property
+    def is_assigned(self) -> bool:
+        return self._assignment is not None
+
 
 class ProcessingElement(Resource):
     """
@@ -210,6 +225,11 @@ class ProcessingElement(Resource):
         self._entity_name = entity_name
         self._input_count = ops[0].input_count
         self._output_count = ops[0].output_count
+        self._assignment = list(
+            self._collection.split_on_execution_time(heuristic="left_edge")
+        )
+        if len(self._assignment) > 1:
+            raise ValueError("Cannot map ProcessCollection to single ProcessingElement")
 
     @property
     def processes(self) -> Set[OperatorProcess]:
@@ -275,6 +295,19 @@ class Memory(Resource):
         # Add information about the iterator type
         return cast(Iterator[MemoryVariable], iter(self._collection))
 
+    def _assign_ram(self, heuristic: str = "left_edge"):
+        """
+        Perform RAM-type assignment of MemoryVariables in this Memory.
+
+        Parameters
+        ----------
+        heuristic : {'left_edge', 'graph_color'}
+            The underlying heuristic to use when performing RAM assignment.
+        """
+        self._assignment = list(
+            self._collection.split_on_execution_time(heuristic=heuristic)
+        )
+
 
 class Architecture(HardwareBlock):
     """

b_asic/resources.py

@@ -852,7 +852,7 @@ class ProcessCollection:
             )
             return self._split_from_graph_coloring(coloring)
         elif heuristic == "left_edge":
-            raise NotImplementedError()
+            return self._left_edge_assignment()
         else:
             raise ValueError(f"Invalid heuristic '{heuristic}'")
 
@@ -989,18 +989,17 @@ class ProcessCollection:
     def __iter__(self):
         return iter(self._collection)
 
-    def graph_color_cell_assignment(
+    def _graph_color_assignment(
         self,
         coloring_strategy: str = "saturation_largest_first",
         *,
         coloring: Optional[Dict[Process, int]] = None,
-    ) -> Set["ProcessCollection"]:
+    ) -> List["ProcessCollection"]:
         """
-        Perform cell assignment of the processes in this collection using graph
-        coloring.
+        Perform assignment of the processes in this collection using graph coloring.
 
-        Two or more processes can share a single cell if, and only if, they have no
-        overlaping time alive.
+        Two or more processes can share a single resource if, and only if, they have no
+        overlaping execution time.
 
         Parameters
         ----------
@@ -1014,7 +1013,7 @@ class ProcessCollection:
 
         Returns
         -------
-        A set of ProcessCollection
+        List[ProcessCollection]
 
         """
         cell_assignment: Dict[int, ProcessCollection] = dict()
@@ -1027,19 +1026,19 @@ class ProcessCollection:
             if cell not in cell_assignment:
                 cell_assignment[cell] = ProcessCollection(set(), self._schedule_time)
             cell_assignment[cell].add_process(process)
-        return set(cell_assignment.values())
+        return list(cell_assignment.values())
 
-    def left_edge_cell_assignment(self) -> Dict[int, "ProcessCollection"]:
+    def _left_edge_assignment(self) -> List["ProcessCollection"]:
         """
-        Perform cell assignment of the processes in this collection using the left-edge
+        Perform assignment of the processes in this collection using the left-edge
         algorithm.
 
-        Two or more processes can share a single cell if, and only if, they have no
-        overlaping time alive.
+        Two or more processes can share a single resource if, and only if, they have no
+        overlaping execution time.
 
         Returns
         -------
-        Dict[int, ProcessCollection]
+        List[ProcessCollection]
         """
         next_empty_cell = 0
         cell_assignment: Dict[int, ProcessCollection] = dict()
@@ -1070,7 +1069,7 @@ class ProcessCollection:
                 )
                 cell_assignment[next_empty_cell].add_process(next_process)
                 next_empty_cell += 1
-        return cell_assignment
+        return [pc for pc in cell_assignment.values()]
 
     def generate_memory_based_storage_vhdl(
         self,

test/test_resources.py

@@ -34,27 +34,30 @@ class TestProcessCollectionPlainMemoryVariable:
     @pytest.mark.mpl_image_compare(style='mpl20')
     def test_left_edge_cell_assignment(self, simple_collection: ProcessCollection):
         fig, ax = plt.subplots(1, 2)
-        assignment = simple_collection.left_edge_cell_assignment()
-        for cell in assignment:
-            assignment[cell].plot(ax=ax[1], row=cell)  # type: ignore
+        assignment = list(simple_collection._left_edge_assignment())
+        for i, cell in enumerate(assignment):
+            cell.plot(ax=ax[1], row=i)  # type: ignore
         simple_collection.plot(ax[0])  # type:ignore
         return fig
 
     def test_cell_assignment_matrix_transposer(self):
         collection = generate_matrix_transposer(4, min_lifetime=5)
-        assignment_left_edge = collection.left_edge_cell_assignment()
-        assignment_graph_color = collection.graph_color_cell_assignment(
+        assignment_left_edge = collection._left_edge_assignment()
+        assignment_graph_color = collection.split_on_execution_time(
             coloring_strategy='saturation_largest_first'
         )
-        assert len(assignment_left_edge.keys()) == 18
+        assert len(assignment_left_edge) == 18
         assert len(assignment_graph_color) == 16
 
     def test_generate_memory_based_vhdl(self):
         for rows in [2, 3, 4, 5, 7]:
             collection = generate_matrix_transposer(rows, min_lifetime=0)
-            assignment = collection.graph_color_cell_assignment()
+            assignment = collection.split_on_execution_time(heuristic="graph_color")
             collection.generate_memory_based_storage_vhdl(
-                filename=f'b_asic/codegen/testbench/streaming_matrix_transposition_memory_{rows}x{rows}.vhdl',
+                filename=(
+                    'b_asic/codegen/testbench/'
+                    f'streaming_matrix_transposition_memory_{rows}x{rows}.vhdl'
+                ),
                 entity_name=f'streaming_matrix_transposition_memory_{rows}x{rows}',
                 assignment=assignment,
                 word_length=16,
@@ -65,22 +68,31 @@ class TestProcessCollectionPlainMemoryVariable:
             generate_matrix_transposer(
                 rows, min_lifetime=0
             ).generate_register_based_storage_vhdl(
-                filename=f'b_asic/codegen/testbench/streaming_matrix_transposition_register_{rows}x{rows}.vhdl',
+                filename=(
+                    'b_asic/codegen/testbench/streaming_matrix_transposition_'
+                    f'register_{rows}x{rows}.vhdl'
+                ),
                 entity_name=f'streaming_matrix_transposition_register_{rows}x{rows}',
                 word_length=16,
             )
 
     def test_rectangular_matrix_transposition(self):
         collection = generate_matrix_transposer(rows=4, cols=8, min_lifetime=2)
-        assignment = collection.graph_color_cell_assignment()
+        assignment = collection.split_on_execution_time(heuristic="graph_color")
         collection.generate_memory_based_storage_vhdl(
-            filename='b_asic/codegen/testbench/streaming_matrix_transposition_memory_4x8.vhdl',
+            filename=(
+                'b_asic/codegen/testbench/streaming_matrix_transposition_memory_'
+                '4x8.vhdl'
+            ),
             entity_name='streaming_matrix_transposition_memory_4x8',
             assignment=assignment,
             word_length=16,
         )
         collection.generate_register_based_storage_vhdl(
-            filename='b_asic/codegen/testbench/streaming_matrix_transposition_register_4x8.vhdl',
+            filename=(
+                'b_asic/codegen/testbench/streaming_matrix_transposition_register_'
+                '4x8.vhdl'
+            ),
             entity_name='streaming_matrix_transposition_register_4x8',
             word_length=16,
         )