Add support for operator processcollections

b_asic/process.py

@@ -29,8 +29,8 @@ class Process:
         self._start_time = start_time
         self._execution_time = execution_time
         if name is None:
-            self._name = f"Proc. {PlainMemoryVariable._name_cnt}"
-            PlainMemoryVariable._name_cnt += 1
+            self._name = f"Proc. {Process._name_cnt}"
+            Process._name_cnt += 1
         else:
             self._name = name
 
@@ -87,12 +87,12 @@ class OperatorProcess(Process):
         execution_time = operation.execution_time
         if execution_time is None:
             raise ValueError(
-                "Operation {operation!r} does not have an execution time specified!"
+                f"Operation {operation!r} does not have an execution time specified!"
             )
         super().__init__(
             start_time,
             execution_time,
-            name=name,
+            name=name or operation.name or operation.graph_id,
         )
         self._operation = operation
 

b_asic/resources.py

@@ -10,6 +10,7 @@ from matplotlib.ticker import MaxNLocator
 
 from b_asic._preferences import LATENCY_COLOR
 from b_asic.process import MemoryVariable, PlainMemoryVariable, Process
+from b_asic.types import TypeName
 
 # Default latency coloring RGB tuple
 _LATENCY_COLOR = tuple(c / 255 for c in LATENCY_COLOR)
@@ -1120,3 +1121,29 @@ class ProcessCollection:
                 write_ports=write_ports,
                 total_ports=total_ports,
             )
+
+    def get_by_type_name(self, type_name: TypeName) -> "ProcessCollection":
+        """
+        Return a ProcessCollection with only a given type of operations.
+
+        Parameters
+        ----------
+        type_name : TypeName
+            The type_name of the operation.
+
+        Returns
+        -------
+        None.
+
+        """
+        from b_asic.process import OperatorProcess
+
+        return ProcessCollection(
+            {
+                process
+                for process in self._collection
+                if isinstance(process, OperatorProcess)
+                and process._operation.type_name() == type_name
+            },
+            self._schedule_time,
+        )

b_asic/schedule.py

@@ -31,7 +31,7 @@ from b_asic._preferences import (
 from b_asic.graph_component import GraphID
 from b_asic.operation import Operation
 from b_asic.port import InputPort, OutputPort
-from b_asic.process import MemoryVariable
+from b_asic.process import MemoryVariable, OperatorProcess
 from b_asic.resources import ProcessCollection
 from b_asic.signal_flow_graph import SFG
 from b_asic.special_operations import Delay, Input, Output
@@ -294,13 +294,13 @@ class Schedule:
 
     @property
     def start_times(self) -> Dict[GraphID, int]:
-        """The start times of the operations in the current schedule."""
+        """The start times of the operations in the schedule."""
         return self._start_times
 
     @property
     def laps(self) -> Dict[GraphID, int]:
         """
-        The number of laps for the start times of the operations in the current schedule.
+        The number of laps for the start times of the operations in the schedule.
         """
         return self._laps
 
@@ -674,8 +674,8 @@ class Schedule:
                 ] + cast(int, source_port.latency_offset)
         self._remove_delays()
 
-    def _get_memory_variables_list(self) -> List['MemoryVariable']:
-        ret: List['MemoryVariable'] = []
+    def _get_memory_variables_list(self) -> List[MemoryVariable]:
+        ret: List[MemoryVariable] = []
         for graph_id, start_time in self._start_times.items():
             slacks = self._forward_slacks(graph_id)
             for outport, signals in slacks.items():
@@ -707,6 +707,24 @@ class Schedule:
             set(self._get_memory_variables_list()), self.schedule_time
         )
 
+    def get_operations(self) -> ProcessCollection:
+        """
+        Return a :class:`~b_asic.resources.ProcessCollection` containing all
+        operations.
+
+        Returns
+        -------
+        ProcessCollection
+
+        """
+        return ProcessCollection(set(self._get_operations_list()), self.schedule_time)
+
+    def _get_operations_list(self) -> List[OperatorProcess]:
+        return [
+            OperatorProcess(start_time, self._sfg.find_by_id(graph_id))
+            for graph_id, start_time in self._start_times.items()
+        ]
+
     def _get_y_position(
         self, graph_id, operation_height=1.0, operation_gap=None
     ) -> float:

examples/secondorderdirectformiir.py

@@ -35,6 +35,8 @@ sfg.set_latency_of_type(ConstantMultiplication.type_name(), 2)
 sfg.set_latency_of_type(Addition.type_name(), 1)
 sfg.set_execution_time_of_type(ConstantMultiplication.type_name(), 1)
 sfg.set_execution_time_of_type(Addition.type_name(), 1)
+sfg.set_execution_time_of_type(Input.type_name(), 1)
+sfg.set_execution_time_of_type(Output.type_name(), 1)
 
 # %%
 # Create schedule