From 0f62a9ff3f7c1fbf2b85bfde649dcd4b34c1a168 Mon Sep 17 00:00:00 2001 From: Adam Glustein Date: Fri, 10 Jul 2026 12:36:12 -0400 Subject: [PATCH] Add optional argument to basketlib sync nodes in order to sync on an external trigger Signed-off-by: Adam Glustein --- cpp/csp/cppnodes/basketlibimpl.cpp | 18 +++-- cpp/csp/python/cspbasketlibimpl.cpp | 2 +- csp/basketlib.py | 38 +++++++---- csp/tests/test_basketlib.py | 71 ++++++++++++++++++-- docs/wiki/api-references/Basket-Nodes-API.md | 11 +-- 5 files changed, 111 insertions(+), 29 deletions(-) diff --git a/cpp/csp/cppnodes/basketlibimpl.cpp b/cpp/csp/cppnodes/basketlibimpl.cpp index 0749a3133..502dd1c1d 100644 --- a/cpp/csp/cppnodes/basketlibimpl.cpp +++ b/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) diff --git a/cpp/csp/python/cspbasketlibimpl.cpp b/cpp/csp/python/cspbasketlibimpl.cpp index c2c59c618..3f9b779b4 100644 --- a/cpp/csp/python/cspbasketlibimpl.cpp +++ b/cpp/csp/python/cspbasketlibimpl.cpp @@ -1,7 +1,7 @@ #include #include -REGISTER_CPPNODE( csp::cppnodes, _sync_list ); +REGISTER_CPPNODE( csp::cppnodes, _sync_list_internal ); REGISTER_CPPNODE( csp::cppnodes, _sample_list ); static PyModuleDef _cspbasketlibimpl_module = { diff --git a/csp/basketlib.py b/csp/basketlib.py index bf0414c05..ed147de3a 100644 --- a/csp/basketlib.py +++ b/csp/basketlib.py @@ -11,10 +11,10 @@ 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) @@ -22,9 +22,11 @@ def sync_list(x: List[ts["T"]], threshold: timedelta, output_incomplete: bool = 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)}") diff --git a/csp/tests/test_basketlib.py b/csp/tests/test_basketlib.py index 275506e6e..de5489a2d 100644 --- a/csp/tests/test_basketlib.py +++ b/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(): diff --git a/docs/wiki/api-references/Basket-Nodes-API.md b/docs/wiki/api-references/Basket-Nodes-API.md index 6c25d964c..33dbd54f5 100644 --- a/docs/wiki/api-references/Basket-Nodes-API.md +++ b/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.