overloaded Integrator with reset times instead of pathsim integrator

Author: RemDelaporteMathurin

src/custom_pathsim_blocks.py

@@ -56,6 +56,44 @@ def __init__(self, f1, f2, f3):
         super().__init__(n=3, fractions=[f1, f2, f3])
 
 
+class Integrator(pathsim.blocks.Integrator):
+    """Integrator block with a reset method."""
+
+    def __init__(self, initial_value=0.0, reset_times=None):
+        """
+        Args:
+            initial_value: Initial value of the integrator.
+            reset_times: List of times at which the integrator is reset. If None, no reset events are created.
+        """
+        super().__init__(initial_value=initial_value)
+        self.reset_times = reset_times
+
+    def create_reset_events(self):
+        """Create reset events for the integrator based on the reset times.
+
+        Raises:
+            ValueError: If reset_times is not valid.
+
+        Returns:
+            list of reset events.
+        """
+        if self.reset_times is None:
+            return []
+        if isinstance(self.reset_times, (int, float)):
+            reset_times = [self.reset_times]
+        elif isinstance(self.reset_times, list) and all(
+            isinstance(t, (int, float)) for t in self.reset_times
+        ):
+            reset_times = self.reset_times
+        else:
+            raise ValueError("reset_times must be a single value or a list of times")
+
+        return [
+            pathsim.blocks.Schedule(t_start=t, t_end=t, func_act=self.reset)
+            for t in reset_times
+        ]
+
+
 # BUBBLER SYSTEM
 
 

src/pathsim_utils.py

@@ -12,7 +12,6 @@
     Amplifier,
     Adder,
     Multiplier,
-    Integrator,
     Function,
     Delay,
     RNG,
@@ -27,6 +26,7 @@
     Splitter3,
     Bubbler,
     FestimWall,
+    Integrator,
 )
 from flask import jsonify
 import inspect
@@ -82,24 +82,13 @@ def create_integrator(
     if eval_namespace is None:
         eval_namespace = globals()
 
-    block = Integrator(
-        initial_value=eval(node["data"]["initial_value"], eval_namespace)
-        if node["data"].get("initial_value") and node["data"]["initial_value"] != ""
-        else 0.0,
+    parameters = get_parameters_for_block_class(
+        Integrator, node, eval_namespace=eval_namespace
     )
+
+    block = Integrator(**parameters)
     # add events to reset integrator if needed
-    events = []
-    if node["data"]["reset_times"] != "":
-
-        def reset_itg(_):
-            block.reset()
-
-        reset_times = eval(node["data"]["reset_times"], eval_namespace)
-        if isinstance(reset_times, (int, float)):
-            # If it's a single number, convert it to a list
-            reset_times = [reset_times]
-        for t in reset_times:
-            events.append(Schedule(t_start=t, t_end=t, func_act=reset_itg))
+    events = block.create_reset_events()
     return block, events
 
 
@@ -161,7 +150,7 @@ def create_bubbler(node: dict) -> Bubbler:
     return block, events
 
 
-def create_scope(node: dict, edges, nodes) -> Scope:
+def make_labels_for_scope(node: dict, edges: list, nodes: list) -> list[str]:
     # Find all incoming edges to this node and sort by source id for consistent ordering
     incoming_edges = [edge for edge in edges if edge["target"] == node["id"]]
     incoming_edges.sort(key=lambda x: x["source"])
@@ -188,12 +177,17 @@ def create_scope(node: dict, edges, nodes) -> Scope:
             if edge["sourceHandle"]:
                 labels[i] += f" ({edge['sourceHandle']})"
 
-    parameters = get_parameters_for_block_class(Scope, node, eval_namespace=globals())
-    # remove labels from parameters, as they are set separately
-    parameters.pop("labels", None)
+    return labels, connections_order
+
+
+def create_scope(node: dict, edges, nodes) -> Scope:
+    block = auto_block_construction(node, eval_namespace=globals())
 
-    block = Scope(labels=labels, **parameters)
+    # override labels + add connections order
+    # TODO this should be done in "make connections"
+    labels, connections_order = make_labels_for_scope(node, edges, nodes)
     block._connections_order = connections_order
+    block.labels = labels
 
     return block
 

src/templates/block_macros.py

@@ -13,18 +13,9 @@
 {% macro create_integrator_block(node) -%}
 {{ create_block(node) }}
 
-{%- if node["data"].get("reset_times") %}
-def reset_{{ node["var_name"] }}(_):
-    {{ node["var_name"] }}.reset()
-
-for t in {{ node["data"].get("reset_times", "[]") }}:
-    events.append(
-        pathsim.events.Schedule(
-            t_start=t,
-            t_end=t,
-            func_act=reset_{{ node["var_name"] }},
-        )
-    )
+{%- if node["data"].get("replacement_times") %}
+events_{{ node["var_name"] }} = {{ node["var_name"] }}.create_reset_events()
+events += events_{{ node["var_name"] }}
 {%- endif %}
 
 {%- endmacro -%}

test/test_backend.py

@@ -3,8 +3,9 @@
     auto_block_construction,
     create_function,
     create_bubbler,
+    create_scope,
 )
-from src.custom_pathsim_blocks import Process, Splitter2, Splitter3, Bubbler
+from src.custom_pathsim_blocks import Process, Splitter2, Splitter3, Bubbler, Integrator
 
 import pathsim.blocks
 
@@ -64,7 +65,10 @@
         "type": "splitter3",
         "data": {"f1": "1/3", "f2": "1/3", "f3": "1/3", "label": "Splitter 3"},
     },
-    "scope": {"type": "scope", "data": {"label": "Scope"}},
+    "scope": {
+        "type": "scope",
+        "data": {"label": "Scope", "sampling_rate": "", "labels": ""},
+    },
     "white_noise": {
         "type": "white_noise",
         "data": {
@@ -127,7 +131,7 @@ def test_create_integrator():
     }
     integrator, events = create_integrator(node)
 
-    assert isinstance(integrator, pathsim.blocks.Integrator)
+    assert isinstance(integrator, Integrator)
     assert integrator.initial_value == 0
     for event in events:
         assert isinstance(event, pathsim.blocks.Schedule)
@@ -214,3 +218,15 @@ def test_create_bubbler():
     }
     block, events = create_bubbler(node)
     assert isinstance(block, Bubbler)
+
+
+def test_make_scope():
+    node = {
+        "id": "7",
+        "type": "scope",
+        "data": {"label": "scope 7", "sampling_rate": "", "labels": "", "t_wait": ""},
+    }
+    block = create_scope(node, edges=[], nodes=[node])
+    assert isinstance(block, pathsim.blocks.Scope)
+    assert block.labels == []
+    assert block.sampling_rate is None

test/test_convert_python.py

@@ -37,6 +37,9 @@
             "type": "scope",
             "data": {
                 "label": "scope_1",
+                "labels": "",
+                "sampling_rate": "",
+                "t_wait": "",
             },
         },
     ],

test/test_files/constant_delay_scope.json

@@ -42,7 +42,10 @@
         "y": 220
       },
       "data": {
-        "label": "scope 2"
+        "label": "scope 2",
+        "labels": "",
+        "sampling_rate": "",
+        "t_wait": ""
       },
       "measured": {
         "width": 120,

test/test_files/same_label.json

@@ -26,7 +26,10 @@
         "y": 400.9936264496624
       },
       "data": {
-        "label": "scope 2"
+        "label": "scope 2",
+        "labels": "",
+        "sampling_rate": "",
+        "t_wait": ""
       },
       "measured": {
         "width": 120,