diff --git a/crates/wingfoil-python/README.md b/crates/wingfoil-python/README.md index 176adf9a5..8ff243fee 100644 --- a/crates/wingfoil-python/README.md +++ b/crates/wingfoil-python/README.md @@ -223,6 +223,7 @@ I/O sources are module-level functions taking the graph first — see | `.fold(init, f)` | Fold into an accumulator seeded from `init`, emitting it after each fold. | | `.reduce(f)` | Like `fold`, but the first value seeds the accumulator. | | `.difference()` | Emit `value - previous` (quiet on the first). | +| `.pairwise()` | Emit `(previous, current)` tuples (quiet on the first); works for non-arithmetic values. | | `.neg()` | Arithmetic negation — Python `-value` / `__neg__` (`5 -> -5`). **Not** a logical `not` (`True -> -1`, not `False`) and **not** a bitwise `~` (`5 -> -5`, not `-6`); for those use `.map(lambda v: not v)` or `.map(lambda v: ~v)`. | | `.bimap(other, f)` | Combine two streams through `f(this, other)`, whenever either ticks. | diff --git a/crates/wingfoil-python/src/graph.rs b/crates/wingfoil-python/src/graph.rs index 2c852b699..e52680f9b 100644 --- a/crates/wingfoil-python/src/graph.rs +++ b/crates/wingfoil-python/src/graph.rs @@ -28,6 +28,7 @@ use std::time::Duration; use anyhow::{Context, Result}; use pyo3::IntoPyObject; use pyo3::prelude::*; +use pyo3::types::PyTuple; use wingfoil::interp::{Builder, Handle, Runner, SlotRef}; use wingfoil::op::{Activation, Ctx, Tick}; use wingfoil::prelude::{Burst, GraphBuilder, SourceOps, Stream, StreamOps, Upstream}; @@ -450,6 +451,37 @@ impl PyStream { self.wrap(self.stream.difference()) } + /// Emit successive `(previous, current)` tuples, staying quiet until a + /// previous value exists. Unlike [`difference`](Self::difference), this + /// works for non-arithmetic Python values, and its tuple output composes + /// directly with [`split`](Self::split). + pub fn pairwise(&self) -> PyStream { + let paired = self.stream.wire(move |b: &mut Builder, h| { + b.register_op1( + h, + "pairwise", + Activation::NONE, + (), + || None::, + move |_cfg: &mut (), previous: &mut Option, value: &PyElement, _ctx| { + let out = match previous.take() { + Some(previous) => Python::attach(|py| -> Result> { + let pair = PyTuple::new(py, [previous.value(), value.value()]) + .map_err(|err| { + anyhow::anyhow!("Python pairwise tuple construction: {err}") + })?; + Ok(Tick::Value(PyElement::new(pair.into_any().unbind()))) + })?, + None => Tick::Quiet, + }; + *previous = Some(value.clone()); + Ok(out) + }, + ) + }); + self.wrap(paired) + } + /// Negate each value **arithmetically** — Python `-value` (`__neg__`), so /// `5 -> -5` and `5.0 -> -5.0`. Exposed to Python as `neg`. /// @@ -1414,6 +1446,26 @@ mod tests { assert_eq!(1, v); // 1,2,3,4 -> deltas 1,1,1 } + #[test] + fn pairwise_is_quiet_until_a_previous_string_exists() { + let g = PyGraph::new(); + let pairs = g + .counter(Duration::from_nanos(100)) + .map(lambda("lambda n: f'v{n}'")) + .pairwise() + .collect(); + run_cycles(&g, 3); + let rows: Vec<(i64, (String, String))> = + Python::attach(|py| pairs.value().value().extract(py).unwrap()); + assert_eq!( + vec![ + (100, ("v1".to_string(), "v2".to_string())), + (200, ("v2".to_string(), "v3".to_string())), + ], + rows + ); + } + #[test] fn neg_arithmetically_negates_an_integer() { let g = PyGraph::new(); diff --git a/crates/wingfoil-python/src/python.rs b/crates/wingfoil-python/src/python.rs index b8a03994c..faeeba8cf 100644 --- a/crates/wingfoil-python/src/python.rs +++ b/crates/wingfoil-python/src/python.rs @@ -206,6 +206,11 @@ impl Stream { Stream(self.0.difference()) } + /// Emit successive `(previous, current)` tuples (quiet on the first). + fn pairwise(&self) -> Stream { + Stream(self.0.pairwise()) + } + /// Negate each value arithmetically: `-value`, i.e. Python `__neg__`. /// `5` becomes `-5`, `5.0` becomes `-5.0`. /// diff --git a/crates/wingfoil-python/tests/test_interop.py b/crates/wingfoil-python/tests/test_interop.py index ed0c01907..9df1ced7a 100644 --- a/crates/wingfoil-python/tests/test_interop.py +++ b/crates/wingfoil-python/tests/test_interop.py @@ -478,6 +478,17 @@ def test_difference_of_counter_is_one(): assert out.value() == 1 # 1,2,3,4 -> deltas 1,1,1 +def test_pairwise_splits_previous_and_current_strings(): + g = wf.Graph() + values = g.counter(period_nanos=100).map(lambda n: f"v{n}") + previous, current = values.pairwise().split() + previous = previous.collect() + current = current.collect() + g.run(cycles=3) + assert previous.value() == [(100, "v1"), (200, "v2")] + assert current.value() == [(100, "v2"), (200, "v3")] + + def test_delay_re_emits_each_value_later(): g = wf.Graph() out = g.counter(period_nanos=100).delay(200).collect()