b_asic/resources.py

 import re
-from typing import Dict, List, Optional, Set, Tuple, Union
+from typing import Dict, Iterable, List, Optional, Set, Tuple, TypeVar, Union
 
 import matplotlib.pyplot as plt
 import networkx as nx
@@ -8,9 +8,16 @@ from matplotlib.ticker import MaxNLocator
 
 from b_asic.process import Process
 
+#
+# Human-intuitive sorting:
+# https://stackoverflow.com/questions/2669059/how-to-sort-alpha-numeric-set-in-python
+#
+# Typing '_T' to help Pyright propagate type-information
+#
+_T = TypeVar('_T')
 
-# From https://stackoverflow.com/questions/2669059/how-to-sort-alpha-numeric-set-in-python
-def _sorted_nicely(to_be_sorted):
+
+def _sorted_nicely(to_be_sorted: Iterable[_T]) -> List[_T]:
     """Sort the given iterable in the way that humans expect."""
     convert = lambda text: int(text) if text.isdigit() else text
     alphanum_key = lambda key: [
@@ -92,7 +99,7 @@ class ProcessCollection:
     ----------
     collection : set of :class:`~b_asic.process.Process` objects
         The Process objects forming this ProcessCollection.
-    schedule_time : int, default: 0
+    schedule_time : int
         Length of the time-axis in the generated graph.
     cyclic : bool, default: False
         If the processes operates cyclically, i.e., if time 0 == time *schedule_time*.
@@ -130,7 +137,7 @@ class ProcessCollection:
         Parameters
         ----------
         ax : :class:`matplotlib.axes.Axes`, optional
-            Matplotlib Axes object to draw this lifetime chart onto. If not provided (i.e., set to None), this will
+            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 on return.
         show_name : bool, default: True
             Show name of all processes in the lifetime chart.
@@ -146,7 +153,7 @@ class ProcessCollection:
         else:
             _ax = ax
 
-        # Draw the lifetime chart
+        # Lifetime chart left and right padding
         PAD_L, PAD_R = 0.05, 0.05
         max_execution_time = max(
             process.execution_time for process in self._collection
@@ -157,12 +164,17 @@ class ProcessCollection:
                 f'Error: Schedule time: {self._schedule_time} < Max execution'
                 f' time: {max_execution_time}'
             )
+
+        # Generate the life-time chart
         for i, process in enumerate(_sorted_nicely(self._collection)):
             bar_start = process.start_time % self._schedule_time
             bar_end = (
                 process.start_time + process.execution_time
             ) % self._schedule_time
-            if bar_end > bar_start:
+            if process.execution_time == 0:
+                # Process has no execution time, draw a tick
+                _ax.scatter(x=bar_start, y=i + 1, marker='X', color='blue')
+            elif bar_end > bar_start:
                 _ax.broken_barh(
                     [(PAD_L + bar_start, bar_end - bar_start - PAD_L - PAD_R)],
                     (i + 0.55, 0.9),

b_asic/signal_flow_graph.py

@@ -565,7 +565,7 @@ class SFG(AbstractOperation):
     @property
     def operations(self) -> List[Operation]:
         """Get all operations of this graph in depth-first order."""
-        return self._operations_dfs_order
+        return list(self._operations_dfs_order)
 
     def find_by_type_name(
         self, type_name: TypeName
@@ -1137,7 +1137,8 @@ class SFG(AbstractOperation):
                         self._components_dfs_order.extend(
                             [new_signal, source.operation]
                         )
-                        self._operations_dfs_order.append(source.operation)
+                        if not source.operation in self._operations_dfs_order:
+                            self._operations_dfs_order.append(source.operation)
 
                     # Check if the signal has not been added before.
                     elif (

test/test_operation.py

@@ -156,7 +156,7 @@ class TestToSfg:
 
         assert but1.evaluate(1, 1)[0] == but1_sfg.evaluate(1, 1)[0]
         assert but1.evaluate(1, 1)[1] == but1_sfg.evaluate(1, 1)[1]
-        assert len(but1_sfg.operations) == 8
+        assert len(but1_sfg.operations) == 6
 
     def test_add_to_sfg(self):
         add1 = Addition()

test/test_sfg.py

@@ -24,6 +24,7 @@ from b_asic.core_operations import (
     SymmetricTwoportAdaptor,
 )
 from b_asic.save_load_structure import python_to_sfg, sfg_to_python
+from b_asic.special_operations import Delay
 
 
 class TestInit:
@@ -1563,6 +1564,27 @@ class TestSFGErrors:
             sfg.inputs_required_for_output(1)
 
 
+class TestInputDuplicationBug:
+    def test_input_is_not_duplicated_in_operation_list(self):
+        # Inputs:
+        in1 = Input(name="in1")
+        out1 = Output(name="out1")
+
+        # Operations:
+        t1 = Delay(initial_value=0, name="")
+        t1.inputs[0].connect(in1)
+        add1 = t1 + in1
+
+        out1.inputs[0].connect(add1)
+
+        twotapfir = SFG(inputs=[in1], outputs=[out1], name='twotapfir')
+
+        assert (
+            len([op for op in twotapfir.operations if isinstance(op, Input)])
+            == 1
+        )
+
+
 class TestCriticalPath:
     def test_single_accumulator(self, sfg_simple_accumulator: SFG):
         sfg_simple_accumulator.set_latency_of_type(Addition.type_name(), 5)