Add optional argument to basketlib sync nodes in order to sync on an external trigger (#721)

Signed-off-by: Adam Glustein <Adam.Glustein@Point72.com>

Author: AdamGlustein

cpp/csp/cppnodes/basketlibimpl.cpp

@@ -4,11 +4,14 @@
 namespace csp::cppnodes
 {
 
-DECLARE_CPPNODE( _sync_list )
+DECLARE_CPPNODE( _sync_list_internal )
 {
     TS_LISTBASKET_INPUT( Generic, x );
+    TS_INPUT( Generic, trigger );
+
     SCALAR_INPUT( TimeDelta, threshold );
     SCALAR_INPUT( bool, output_incomplete );
+    SCALAR_INPUT( bool, use_trigger );
 
     ALARM( bool, a_end );
 
@@ -18,7 +21,7 @@ DECLARE_CPPNODE( _sync_list )
 
     TS_LISTBASKET_OUTPUT( Generic );
 
-    INIT_CPPNODE( _sync_list ) { }
+    INIT_CPPNODE( _sync_list_internal ) { }
 
     START()
     {
@@ -27,13 +30,14 @@ DECLARE_CPPNODE( _sync_list )
 
     INVOKE()
     {
-        if( x.tickedinputs() )
+        if( s_alarm_handle.expired() )
         {
-            if( s_alarm_handle.expired() )
-            {
+            if( !use_trigger || trigger.ticked() )
                 s_alarm_handle = csp.schedule_alarm( a_end, threshold, true );
-            }
+        }
 
+        if( s_alarm_handle.active() && x.tickedinputs() )
+        {
             for( auto it = x.tickedinputs(); it; ++it )
             {
                 if( s_current_ticked[ it.elemId() ] == false )
@@ -70,7 +74,7 @@ DECLARE_CPPNODE( _sync_list )
     }
 };
 
-EXPORT_CPPNODE( _sync_list );
+EXPORT_CPPNODE( _sync_list_internal );
 
 /*
 @csp.node(cppimpl=_cspbasketlibimpl._sample_list)

cpp/csp/python/cspbasketlibimpl.cpp

@@ -1,7 +1,7 @@
 #include <csp/engine/CppNode.h>
 #include <csp/python/PyCppNode.h>
 
-REGISTER_CPPNODE( csp::cppnodes, _sync_list );
+REGISTER_CPPNODE( csp::cppnodes, _sync_list_internal );
 REGISTER_CPPNODE( csp::cppnodes, _sample_list );
 
 static PyModuleDef _cspbasketlibimpl_module = {

csp/basketlib.py

@@ -11,20 +11,22 @@
 Y = TypeVar("Y")
 
 
-@csp.node(cppimpl=_cspbasketlibimpl._sync_list)
-def sync_list(x: List[ts["T"]], threshold: timedelta, output_incomplete: bool = True) -> csp.OutputBasket(
-    List[ts["T"]], shape_of="x"
-):
+@csp.node(cppimpl=_cspbasketlibimpl._sync_list_internal)
+def sync_list_internal(
+    x: List[ts["T"]], trigger: ts["K"], threshold: timedelta, output_incomplete: bool, use_trigger: bool
+) -> csp.OutputBasket(List[ts["T"]], shape_of="x"):
     with csp.alarms():
         a_end = csp.alarm(bool)
 
     with csp.state():
         s_current = {}
         s_alarm_handle = None
 
-    if csp.ticked(x):
-        if not s_alarm_handle:
+    if not s_alarm_handle:
+        if not use_trigger or csp.ticked(trigger):
             s_alarm_handle = csp.schedule_alarm(a_end, threshold, True)
+
+    if s_alarm_handle and csp.ticked(x):
         s_current.update(x.tickeditems())
 
     if csp.ticked(a_end) or len(s_current) == len(x):
@@ -37,19 +39,29 @@ def sync_list(x: List[ts["T"]], threshold: timedelta, output_incomplete: bool =
 
 
 @csp.graph
-def sync_dict(x: Dict["K", ts["T"]], threshold: timedelta, output_incomplete: bool = True) -> csp.OutputBasket(
-    Dict["K", ts["T"]], shape_of="x"
-):
+def sync_list(
+    x: List[ts["T"]], threshold: timedelta, output_incomplete: bool = True, trigger: ts["K"] = None
+) -> csp.OutputBasket(List[ts["T"]], shape_of="x"):
+    use_trigger = trigger is not None
+    if not use_trigger:
+        trigger = csp.null_ts(bool)
+    return sync_list_internal(x, trigger, threshold, output_incomplete, use_trigger)
+
+
+@csp.graph
+def sync_dict(
+    x: Dict["K", ts["T"]], threshold: timedelta, output_incomplete: bool = True, trigger: ts["K"] = None
+) -> csp.OutputBasket(Dict["K", ts["T"]], shape_of="x"):
     values = list(x.values())
-    synced = sync_list(values, threshold, output_incomplete)
+    synced = sync_list(values, threshold, output_incomplete, trigger)
     return {k: v for k, v in zip(x.keys(), synced)}
 
 
-def sync(x, threshold: timedelta, output_incomplete: bool = True):
+def sync(x, threshold: timedelta, output_incomplete: bool = True, trigger: ts["K"] = None):
     if isinstance(x, list):
-        return sync_list(x, threshold, output_incomplete)
+        return sync_list(x, threshold, output_incomplete, trigger)
     elif isinstance(x, dict):
-        return sync_dict(x, threshold, output_incomplete)
+        return sync_dict(x, threshold, output_incomplete, trigger)
     raise ValueError(f"Input must be list or dict basket, got: {type(x)}")
 
 

csp/tests/test_basketlib.py

@@ -28,11 +28,19 @@ def test_graph():
             random_floats_async = random_gen(trigger2)
 
             # test _basket_synchronize_list
-            synced_py = basketlib.sync_list.python(
-                x=[random_floats1, random_floats_async], threshold=self.sync_threshold, output_incomplete=True
+            synced_py = basketlib.sync_list_internal.python(
+                x=[random_floats1, random_floats_async],
+                trigger=csp.null_ts(bool),
+                threshold=self.sync_threshold,
+                output_incomplete=True,
+                use_trigger=False,
             )
-            synced_cpp = basketlib.sync_list(
-                x=[random_floats1, random_floats_async], threshold=self.sync_threshold, output_incomplete=True
+            synced_cpp = basketlib.sync_list_internal(
+                x=[random_floats1, random_floats_async],
+                trigger=csp.null_ts(bool),
+                threshold=self.sync_threshold,
+                output_incomplete=True,
+                use_trigger=False,
             )
 
             synced_auto_list = basketlib.sync(
@@ -132,6 +140,61 @@ def basic_graph():
         self.assertEqual(results["synced_complete_1"], [(datetime(2022, 6, 17, 9, 50), 6.0)])
         self.assertEqual(results["synced_complete_2"], [(datetime(2022, 6, 17, 9, 50), 9.0)])
 
+    def test_sync_basket_with_trigger(self):
+        st = datetime(2020, 1, 1)
+
+        @csp.graph
+        def graph():
+            trigger = csp.curve(
+                typ=str,
+                data=[
+                    (st + timedelta(seconds=10), "trigger-1"),  # regular case
+                    (st + timedelta(seconds=30), "trigger-2"),  # exact same tick as a, includes it
+                    (st + timedelta(seconds=31), "trigger-ignored"),  # in an active period, ignored
+                    (st + timedelta(seconds=50), "trigger-3"),  # only a will get a tick here
+                ],
+            )
+            a = csp.curve(
+                typ=float,
+                data=[
+                    (st + timedelta(seconds=1), 1.0),
+                    (st + timedelta(seconds=12), 2.0),
+                    (st + timedelta(seconds=30), 3.0),
+                    (st + timedelta(seconds=52), 4.0),
+                ],
+            )
+            b = csp.curve(
+                typ=float,
+                data=[
+                    (st + timedelta(seconds=2), 5.0),
+                    (st + timedelta(seconds=14), 6.0),
+                    (st + timedelta(seconds=32), 7.0),
+                    (st + timedelta(seconds=56), 8.0),
+                ],
+            )
+
+            synced_list = basketlib.sync_list(x=[a, b], threshold=timedelta(seconds=5), trigger=trigger)
+            synced_dict = basketlib.sync_dict(x={"a": a, "b": b}, threshold=timedelta(seconds=5), trigger=trigger)
+
+            csp.add_graph_output("list_a", synced_list[0])
+            csp.add_graph_output("list_b", synced_list[1])
+            csp.add_graph_output("dict_a", synced_dict["a"])
+            csp.add_graph_output("dict_b", synced_dict["b"])
+
+        result = csp.run(graph, starttime=st, endtime=st + timedelta(minutes=1))
+
+        expected_a = [
+            (st + timedelta(seconds=14), 2.0),
+            (st + timedelta(seconds=32), 3.0),
+            (st + timedelta(seconds=55), 4.0),
+        ]
+        expected_b = [(st + timedelta(seconds=14), 6.0), (st + timedelta(seconds=32), 7.0)]
+        print(result["list_a"])
+        self.assertEqual(result["list_a"], expected_a)
+        self.assertEqual(result["list_b"], expected_b)
+        self.assertEqual(result["dict_a"], expected_a)
+        self.assertEqual(result["dict_b"], expected_b)
+
     def test_sample_dict_basket(self):
         @csp.graph
         def graph():

docs/wiki/api-references/Basket-Nodes-API.md

@@ -13,25 +13,27 @@ These functions are found in the `csp.basketlib` module and can be called using
 ## `sync_list`
 
 ```python
-sync_list(x: List[ts["T"]], threshold: timedelta, output_incomplete: bool = True) → csp.OutputBasket(
+sync_list(x: List[ts["T"]], threshold: timedelta, output_incomplete: bool = True, trigger: ts["K"] = None) → csp.OutputBasket(
     List[ts["T"]], shape_of="x"
 )
 ```
 
 Synchronizes a list basket of time series within some threshold.
 
-When any element of `x` first ticks, we wait up to `threshold` time for other elements to tick. Once all elements of the list basket tick at least once *or* the threshold elapses and `output_incomplete=True`, we return a list basket with the most recent value of each time series (between the interval's first tick and now) and reset the synchronization interval.
+If `trigger` is specified, then it will control when we begin the synchronization intervals. If it is not specified, any element of `x` will trigger the start of a synchronization interval.
+Once an interval is triggered, we wait up to `threshold` time for all elements of the basket `x` to tick. Once all elements of the list basket tick at least once *or* the threshold elapses and `output_incomplete=True`, we return a list basket with the most recent value of each time series (between the interval's first tick and now) and reset the synchronization interval.
 
 Args:
 
 - **`x`**: a list basket of time series to synchronize.
 - **`threshold`**: the time to wait for all elements of the basket to tick before propagating the values.
 - **`output_incomplete`**: if True, return an incomplete output basket if the threshold elapses before all values tick. Else, do not output in this situation.
+- **`trigger`**: an optional time-series which will trigger a synchronization period. If trigger is used, the first tick of `x` will not trigger a synchronization period.
 
 ## `sync_dict`
 
 ```python
-sync_dict(x: Dict["K", ts["T"]], threshold: timedelta, output_incomplete: bool = True) → csp.OutputBasket(
+sync_dict(x: Dict["K", ts["T"]], threshold: timedelta, output_incomplete: bool = True, trigger: ts["K"] = None) → csp.OutputBasket(
     Dict["K", ts["T"]], shape_of="x"
 )
 ```
@@ -43,11 +45,12 @@ Args:
 - **`x`**: a dict basket of time series to synchronize.
 - **`threshold`**: the time to wait for all elements of the basket to tick before propagating the values.
 - **`output_incomplete`**: if True, return an incomplete output basket if the threshold elapses before all values tick. Else, do not output in this situation.
+- **`trigger`**: an optional time-series which will trigger a synchronization period. If trigger is used, the first tick of `x` will not trigger a synchronization period.
 
 ## `sync`
 
 ```python
-sync(x, threshold: timedelta, output_incomplete: bool = True)
+sync(x, threshold: timedelta, output_incomplete: bool = True, trigger: ts["K"] = None)
 ```
 
 Helper function which calls `sync_list` if x is a list basket and `sync_dict` if x is a dict basket. If x is not a valid basket, it will raise an exception.