Add architecture stub classes

b_asic/architecture.py

+"""
+B-ASIC architecture classes.
+"""
+from typing import Set, cast
+
+from b_asic.process import MemoryVariable, OperatorProcess, PlainMemoryVariable
+from b_asic.resources import ProcessCollection
+
+
+class ProcessingElement:
+    """
+    Create a processing element for a ProcessCollection with OperatorProcesses.
+
+    Parameters
+    ----------
+    process_collection : :class:`~b_asic.resources.ProcessCollection`
+    """
+
+    def __init__(self, process_collection: ProcessCollection):
+        if not len(ProcessCollection):
+            raise ValueError(
+                "Do not create ProcessingElement with empty ProcessCollection"
+            )
+        if not all(
+            isinstance(operator, OperatorProcess)
+            for operator in process_collection.collection
+        ):
+            raise TypeError(
+                "Can only have OperatorProcesses in ProcessCollection when creating"
+                " ProcessingElement"
+            )
+        ops = [
+            cast(operand, OperatorProcess).operation
+            for operand in process_collection.collection
+        ]
+        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()
+
+    def write_code(self, path: str, entity_name: str) -> None:
+        """
+        Write VHDL code for processing element.
+
+        Parameters
+        ----------
+        path : str
+            Directory to write code in.
+        entity_name : str
+        """
+        raise NotImplementedError
+
+
+class Memory:
+    """
+    Create a memory from a ProcessCollection with memory variables.
+
+    Parameters
+    ----------
+    process_collection : :class:`~b_asic.resources.ProcessCollection`
+        The ProcessCollection to create a Memory for.
+    memory_type : {'RAM', 'register'}
+        The type of memory.
+    """
+
+    def __init__(self, process_collection: ProcessCollection, memory_type: str = "RAM"):
+        if not len(ProcessCollection):
+            raise ValueError("Do not create Memory with empty ProcessCollection")
+        if not all(
+            isinstance(operator, (MemoryVariable, PlainMemoryVariable))
+            for operator in process_collection.collection
+        ):
+            raise TypeError(
+                "Can only have MemoryVariable or PlainMemoryVariable in"
+                " ProcessCollection when creating Memory"
+            )
+        self._collection = process_collection
+        self._memory_type = memory_type
+
+    def write_code(self, path: str, entity_name: str) -> None:
+        """
+        Write VHDL code for memory.
+
+        Parameters
+        ----------
+        path : str
+            Directory to write code in.
+        entity_name : str
+
+        Returns
+        -------
+
+
+        """
+        raise NotImplementedError
+
+
+class Architecture:
+    """
+    Class representing an architecture.
+
+    Parameters
+    ----------
+    processing_elements : set of :class:`~b_asic.architecture.ProcessingElement`
+        The processing elements in the architecture.
+    memories : set of :class:`~b_asic.architecture.Memory`
+        The memories in the architecture.
+    name : str, default: "arch"
+        Name for the top-level architecture. Used for the entity and as prefix for all
+        building blocks.
+    """
+
+    def __init__(
+        self,
+        processing_elements: Set[ProcessingElement],
+        memories: Set[Memory],
+        name: str = "arch",
+    ):
+        self._processing_elements = processing_elements
+        self._memories = memories
+        self._name = name
+
+    def write_code(self, path: str) -> None:
+        """
+        Write HDL of architecture.
+
+        Parameters
+        ----------
+        path : str
+            Directory to write code in.
+        """
+        raise NotImplementedError

b_asic/process.py

@@ -96,6 +96,14 @@ class OperatorProcess(Process):
         )
         self._operation = operation
 
+    @property
+    def operation(self) -> Operation:
+        """The Operation that the OperatorProcess corresponds to."""
+        return self._operation
+
+    def __repr__(self) -> str:
+        return f"OperatorProcess({self.start_time}, {self.operation}, {self.name!r})"
+
 
 class MemoryVariable(Process):
     """

b_asic/resources.py

@@ -1001,9 +1001,7 @@ class ProcessCollection:
         for collection in assignment:
             for mv in collection:
                 if mv not in self:
-                    raise ValueError(
-                        f'{mv.__repr__()} is not part of {self.__repr__()}.'
-                    )
+                    raise ValueError(f'{mv!r} is not part of {self!r}.')
 
         # Make sure that concurrent reads/writes do not surpass the port setting
         for mv in self:
@@ -1141,7 +1139,7 @@ class ProcessCollection:
                 process
                 for process in self._collection
                 if isinstance(process, OperatorProcess)
-                and process._operation.type_name() == type_name
+                and process.operation.type_name() == type_name
             },
             self._schedule_time,
             self._cyclic,

docs_sphinx/api/architecture.rst

+***********************
+``b_asic.architecture``
+***********************
+
+.. automodule:: b_asic.architecture
+   :members:
+   :undoc-members:

docs_sphinx/api/index.rst

@@ -5,6 +5,7 @@ API
 .. toctree::
     :maxdepth: 1
 
+    architecture.rst
     core_operations.rst
     graph_component.rst
     operation.rst

test/fixtures/schedule.py

@@ -11,3 +11,16 @@ def secondorder_iir_schedule(precedence_sfg_delays):
 
     schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
     return schedule
+
+
+@pytest.fixture
+def secondorder_iir_schedule_with_execution_times(precedence_sfg_delays):
+    precedence_sfg_delays.set_latency_of_type(Addition.type_name(), 4)
+    precedence_sfg_delays.set_latency_of_type(ConstantMultiplication.type_name(), 3)
+    precedence_sfg_delays.set_execution_time_of_type(Addition.type_name(), 2)
+    precedence_sfg_delays.set_execution_time_of_type(
+        ConstantMultiplication.type_name(), 1
+    )
+
+    schedule = Schedule(precedence_sfg_delays, scheduling_algorithm="ASAP")
+    return schedule

test/test_process.py

 import re
 
+import pytest
+
 from b_asic.process import PlainMemoryVariable
 
 
@@ -32,3 +34,8 @@ def test_MemoryVariables(secondorder_iir_schedule):
     assert pattern.match(repr(mem_var))
     assert mem_var.execution_time == 4
     assert mem_var.start_time == 3
+
+
+def test_OperatorProcess_error(secondorder_iir_schedule):
+    with pytest.raises(ValueError, match="does not have an execution time specified"):
+        _ = secondorder_iir_schedule.get_operations()

test/test_resources.py

@@ -4,6 +4,7 @@ import re
 import matplotlib.pyplot as plt
 import pytest
 
+from b_asic.core_operations import ConstantMultiplication
 from b_asic.process import PlainMemoryVariable
 from b_asic.research.interleaver import (
     generate_matrix_transposer,
@@ -124,3 +125,12 @@ class TestProcessCollectionPlainMemoryVariable:
 
     def test_len_process_collection(self, simple_collection: ProcessCollection):
         assert len(simple_collection) == 7
+
+    def test_get_by_type_name(self, secondorder_iir_schedule_with_execution_times):
+        pc = secondorder_iir_schedule_with_execution_times.get_operations()
+        pc_cmul = pc.get_by_type_name(ConstantMultiplication.type_name())
+        assert len(pc_cmul) == 7
+        assert all(
+            isinstance(operand.operation, ConstantMultiplication)
+            for operand in pc_cmul.collection
+        )

test/test_schedule.py

@@ -6,6 +6,7 @@ import re
 import pytest
 
 from b_asic.core_operations import Addition, Butterfly, ConstantMultiplication
+from b_asic.process import OperatorProcess
 from b_asic.schedule import Schedule
 from b_asic.signal_flow_graph import SFG
 from b_asic.special_operations import Delay, Input, Output
@@ -562,3 +563,8 @@ class TestErrors:
             NotImplementedError, match="No algorithm with name: foo defined."
         ):
             Schedule(sfg_simple_filter, scheduling_algorithm="foo")
+
+    def test_get_operations(self, secondorder_iir_schedule_with_execution_times):
+        pc = secondorder_iir_schedule_with_execution_times.get_operations()
+        assert len(pc) == 13
+        assert all(isinstance(operand, OperatorProcess) for operand in pc.collection)