wingfoil-python 8.0.0

use log::Level;
use pyo3::BoundObject;
use std::any::type_name;

use ::wingfoil::adapters::statistics::{StatisticsOperators, Weighting, Window};
use ::wingfoil::{Element, IntoStream, NodeOperators, Stream, StreamOperators};

use pyo3::conversion::IntoPyObject;
use pyo3::prelude::*;

use std::rc::Rc;

use crate::proxy_stream::*;
use crate::py_element::PyElement;
use crate::types::*;
use crate::*;

/// Wrap a Python exception raised inside a user callback into an `anyhow::Error`
/// so it can flow through the fallible (`try_*`) graph operators. The original
/// `PyErr` is recovered by [`ToPyResult`] at the `#[pymethods]` boundary, so the
/// Python type, message and traceback propagate unchanged.
pub(crate) fn py_callback_error(err: PyErr) -> anyhow::Error {
    anyhow::Error::new(err)
}

#[derive(Clone)]
#[pyclass(subclass, unsendable, name = "Stream", from_py_object)]
pub struct PyStream(pub Rc<dyn Stream<PyElement>>);

impl PyStream {
    fn extract<T>(&self) -> Rc<dyn Stream<T>>
    where
        T: Element + for<'a, 'py> FromPyObject<'a, 'py>,
    {
        self.0.try_map(move |x: PyElement| {
            Python::attach(|py| {
                x.as_ref().extract::<T>(py).map_err(|e| {
                    py_callback_error(e.into()).context(format!(
                        "failed to convert Python value to native {}",
                        type_name::<T>()
                    ))
                })
            })
        })
    }

    pub fn inner_stream(&self) -> Rc<dyn Stream<PyElement>> {
        self.0.clone()
    }

    pub fn from_inner(inner: Rc<dyn Stream<PyElement>>) -> Self {
        Self(inner)
    }
}

/// Convert a native Rust value into a `Py<PyAny>`.
///
/// Only ever called with concrete types (`f64`, `u64`, `Vec<Py<PyAny>>`) whose
/// `IntoPyObject` impl is infallible, so a failure here is an invariant
/// violation rather than a user-facing error.
pub fn to_pyany<T>(x: T) -> Py<PyAny>
where
    T: for<'py> IntoPyObject<'py>,
{
    Python::attach(|py| {
        x.into_pyobject(py)
            .map(|bound| bound.into_any().unbind())
            .unwrap_or_else(|_| {
                panic!(
                    "invariant: IntoPyObject for {} is expected to be infallible",
                    type_name::<T>()
                )
            })
    })
}

pub fn vec_any_to_pyany(x: Vec<Py<PyAny>>) -> Py<PyAny> {
    Python::attach(|py| {
        x.into_pyobject(py)
            .map(|bound| bound.into_any().unbind())
            .expect("invariant: IntoPyObject for Vec<Py<PyAny>> is infallible")
    })
}

pub trait AsPyStream<T>
where
    T: Element + for<'py> IntoPyObject<'py>,
{
    fn as_py_stream(&self) -> PyStream;
}

impl<T> AsPyStream<T> for Rc<dyn Stream<T>>
where
    T: Element + for<'py> IntoPyObject<'py>,
{
    fn as_py_stream(&self) -> PyStream {
        let strm = self.map(|x| {
            let py_any = to_pyany(x);
            PyElement::new(py_any)
        });
        PyStream(strm)
    }
}

#[pymethods]
impl PyStream {
    #[new]
    fn new(inner: Py<PyAny>) -> Self {
        let stream = PyProxyStream::new(inner);
        let stream = stream.into_stream();
        Self(stream)
    }

    #[pyo3(signature = (realtime, start=None, duration=None, cycles=None))]
    fn run(
        &self,
        py: Python<'_>,
        realtime: bool,
        start: Option<Py<PyAny>>,
        duration: Option<Py<PyAny>>,
        cycles: Option<u32>,
    ) -> PyResult<()> {
        let (run_mode, run_for) =
            parse_run_args(py, realtime, start, duration, cycles).to_pyresult()?;

        // Convert fat pointer to (addr, vtable) pair which is Send+Sync
        let stream_ptr = Rc::as_ptr(&self.0);
        let (addr, vtable): (usize, usize) = unsafe { std::mem::transmute(stream_ptr) };

        // Release GIL during the run to allow async tasks to acquire it
        // SAFETY: The Rc is kept alive by self for the duration of this call
        let result = py.detach(move || {
            // Reconstruct the fat pointer from (addr, vtable)
            let stream_ptr: *const dyn Stream<PyElement> =
                unsafe { std::mem::transmute((addr, vtable)) };
            // Temporarily reconstruct the Rc without taking ownership
            let stream = unsafe { Rc::from_raw(stream_ptr) };
            let result = stream.run(run_mode, run_for);
            std::mem::forget(stream); // Don't drop the Rc (self.0 still owns it)
            result
        });
        result.to_pyresult()?;
        Ok(())
    }

    fn peek_value(&self) -> Py<PyAny> {
        self.0.peek_value().value()
    }

    // begin StreamOperators

    fn collect(&self) -> PyStream {
        let strm = self.0.collect().map(|items| {
            Python::attach(move |py| {
                let items = items
                    .iter()
                    .map(|item| item.value.as_ref().clone_ref(py))
                    .collect::<Vec<_>>();
                PyElement::new(vec_any_to_pyany(items))
            })
        });
        PyStream(strm)
    }

    fn dataframe(&self) -> PyStream {
        let time_stream = self.0.clone().as_node().ticked_at_elapsed();

        let zipped = ::wingfoil::bimap(
            Dep::Active(self.0.clone()),
            Dep::Active(time_stream),
            |val: PyElement, time: ::wingfoil::NanoTime| {
                Python::attach(|py| {
                    let time_secs: f64 = time.into();

                    let py_tuple = pyo3::types::PyTuple::new(
                        py,
                        &[
                            time_secs
                                .into_pyobject(py)
                                .expect("invariant: IntoPyObject for f64 is infallible")
                                .into_any(),
                            val.value().into_bound(py),
                        ],
                    )
                    .expect("invariant: fixed-size tuple construction cannot fail");

                    PyElement::new(py_tuple.into_any().unbind())
                })
            },
        );

        let strm = zipped.collect().map(|items| {
            Python::attach(move |py| {
                let items = items
                    .iter()
                    .map(|item| item.value.as_ref().clone_ref(py))
                    .collect::<Vec<_>>();
                PyElement::new(vec_any_to_pyany(items))
            })
        });

        PyStream(strm)
    }

    fn average(&self) -> PyStream {
        self.extract::<f64>()
            .mean(Window::Unbounded, Weighting::Count)
            .as_py_stream()
    }

    fn buffer(&self, capacity: usize) -> PyStream {
        let strm = self.0.buffer(capacity).map(|items| {
            Python::attach(move |py| {
                let items = items
                    .iter()
                    .map(|item| item.as_ref().clone_ref(py))
                    .collect::<Vec<_>>();
                PyElement::new(vec_any_to_pyany(items))
            })
        });
        PyStream(strm)
    }

    fn finally(&self, func: Py<PyAny>) -> PyNode {
        let node = self.0.finally(move |py_elmnt, _| {
            Python::attach(move |py| {
                let res = py_elmnt.as_ref().clone_ref(py);
                let args = (res,);
                func.call1(py, args).map_err(py_callback_error)?;
                Ok(())
            })
        });
        PyNode(node)
    }

    fn for_each(&self, func: Py<PyAny>) -> PyNode {
        let node = self.0.try_for_each(move |py_elmnt, t| {
            Python::attach(|py| {
                let res = py_elmnt.as_ref().clone_ref(py);
                let t: f64 = t.into();
                let args = (res, t);
                func.call1(py, args).map_err(py_callback_error)?;
                Ok(())
            })
        });
        PyNode(node)
    }

    fn inspect(&self, func: Py<PyAny>) -> PyStream {
        // No fallible `inspect` operator exists in core; express the
        // side-effecting pass-through as a `try_map` that returns the value
        // unchanged so a Python exception in `func` propagates.
        let stream = self.0.try_map(move |x: PyElement| {
            Python::attach(|py| {
                func.call1(py, (x.value(),)).map_err(py_callback_error)?;
                Ok(x)
            })
        });
        PyStream(stream)
    }

    /// difference in its source from one cycle to the next (pass-through of PyElement)
    fn difference(&self) -> PyStream {
        PyStream(self.0.difference())
    }

    /// Propagates its source delayed by specified duration (milliseconds)
    fn delay(&self, delay_secs: f64) -> PyStream {
        let delay = Duration::from_secs_f64(delay_secs);
        PyStream(self.0.delay(delay))
    }

    /// only propagates its source if it changed (uses PartialEq on PyElement)
    fn distinct(&self) -> PyStream {
        PyStream(self.0.distinct())
    }

    /// drops source contingent on supplied predicate (Python callable)
    fn filter(&self, keep_func: Py<PyAny>) -> PyStream {
        let keep = self.0.try_map(move |x: PyElement| {
            Python::attach(|py| {
                let res = keep_func
                    .call1(py, (x.value(),))
                    .map_err(py_callback_error)?;
                res.extract::<bool>(py).map_err(|e| {
                    py_callback_error(e).context("filter predicate must return a bool")
                })
            })
        });
        PyStream(self.0.filter(keep))
    }

    /// propagates source up to limit times
    fn limit(&self, limit: u32) -> PyStream {
        PyStream(self.0.limit(limit))
    }

    /// logs source and propagates it. Default level INFO.
    fn logged(&self, label: String) -> PyStream {
        PyStream(self.0.logged(&label, Level::Info))
    }

    /// Map’s its source into a new Stream using the supplied Python callable.
    fn map(&self, func: Py<PyAny>) -> PyStream {
        let stream = self.0.try_map(move |x: PyElement| {
            Python::attach(|py| {
                let res = func.call1(py, (x.value(),)).map_err(py_callback_error)?;
                Ok(PyElement::new(res))
            })
        });
        PyStream(stream)
    }

    // /// negates its input (for boolean-like PyElements)
    fn not(&self) -> PyStream {
        PyStream(self.0.not())
    }

    fn sample(&self, trigger: Py<PyAny>) -> PyResult<PyStream> {
        Python::attach(|py| {
            let obj = trigger.as_ref();
            if let Ok(node) = obj.extract::<PyRef<PyNode>>(py) {
                return Ok(PyStream(self.0.sample(node.0.clone())));
            }
            if let Ok(stream) = obj.extract::<PyRef<PyStream>>(py) {
                return Ok(PyStream(self.0.sample(stream.0.clone())));
            }
            Err::<PyStream, _>(py_type_error("sample: expected a Node or Stream trigger"))
                .to_pyresult()
        })
    }

    /// sum the stream (extracts f64 values before summing)
    fn sum(&self) -> PyStream {
        self.extract::<f64>().sum(Window::Unbounded).as_py_stream()
    }

    fn count(&self) -> PyStream {
        self.0.count().as_py_stream()
    }

    /// Pairs each value with the graph time as a `(float, value)` tuple,
    /// where the float is seconds since Unix epoch.
    fn with_time(&self) -> PyStream {
        let strm = self.0.with_time().map(|(t, v)| {
            Python::attach(|py| {
                let time_secs: f64 = t.into();
                let py_tuple = pyo3::types::PyTuple::new(
                    py,
                    &[
                        time_secs
                            .into_pyobject(py)
                            .expect("invariant: IntoPyObject for f64 is infallible")
                            .into_any(),
                        v.value().into_bound(py),
                    ],
                )
                .expect("invariant: fixed-size tuple construction cannot fail");
                PyElement::new(py_tuple.into_any().unbind())
            })
        });
        PyStream(strm)
    }

    /// Write this stream of dicts to a CSV file.
    ///
    /// Each dict becomes one CSV row. Headers are inferred from the first dict's
    /// keys, and a `time` column is prepended with the graph time in nanoseconds.
    ///
    /// Args:
    ///     path: Output file path
    ///
    /// Returns:
    ///     A Node that drives the write operation.
    fn csv_write(&self, path: String) -> PyResult<PyNode> {
        let node = crate::py_csv::py_csv_write_inner(&self.0, path).to_pyresult()?;
        Ok(PyNode::new(node))
    }

    /// Forecast this stream of floats with an augurs model.
    ///
    /// Buffers a sliding window of the last `window` values and, once
    /// `min_points` have arrived, emits a dict each tick:
    /// `{"point": list[float], "lower": list[float], "upper": list[float]}`.
    /// `lower`/`upper` are empty unless a `level` (0..1) is requested.
    ///
    /// Uses non-seasonal ETS by default. Pass `periods` (a list of seasonal
    /// period lengths, e.g. `[24]`) to fit an MSTL seasonal model instead.
    ///
    /// Args:
    ///     window: Maximum number of recent points retained as history.
    ///     horizon: Number of steps to forecast ahead.
    ///     level: Confidence level for prediction intervals (0..1), or None.
    ///     min_points: Minimum points before fitting begins. Raised to the
    ///         model's warm-up floor: 12 for ETS, or `2 * max(periods) + 1`
    ///         for MSTL (STL needs two full seasonal periods).
    ///     periods: Seasonal period lengths for MSTL, or None for ETS.
    ///
    /// Raises:
    ///     ValueError: if `level` is not in `(0, 1)`, or any `periods` entry
    ///         is below 2.
    #[pyo3(signature = (window, horizon, level=None, min_points=12, periods=None))]
    fn augurs_forecast(
        &self,
        window: usize,
        horizon: usize,
        level: Option<f64>,
        min_points: usize,
        periods: Option<Vec<usize>>,
    ) -> PyResult<PyStream> {
        Ok(PyStream(crate::py_augurs::py_augurs_forecast_inner(
            &self.0, window, horizon, level, min_points, periods,
        )?))
    }

    /// Detect outlying series in this stream of per-series readings.
    ///
    /// Each value must be a `list[float]` carrying one reading per series.
    /// Buffers the last `window` ticks and emits a dict each tick:
    /// `{"outlying": list[int], "scores": list[float]}`.
    ///
    /// Args:
    ///     window: Number of recent samples in the detection window.
    ///     sensitivity: Detector sensitivity, strictly between 0 and 1.
    ///     detector: "mad" (default) or "dbscan". DBSCAN needs >= 3 series.
    ///
    /// Raises:
    ///     ValueError: if `sensitivity` is not in `(0, 1)`, or `detector` is
    ///         not "mad" or "dbscan".
    #[pyo3(signature = (window, sensitivity, detector="mad"))]
    fn augurs_outlier(
        &self,
        window: usize,
        sensitivity: f64,
        detector: &str,
    ) -> PyResult<PyStream> {
        Ok(PyStream(crate::py_augurs::py_augurs_outlier_inner(
            &self.0,
            window,
            sensitivity,
            detector,
        )?))
    }

    /// Detect changepoints in this stream of floats (Bayesian online
    /// changepoint detection over a sliding window).
    ///
    /// Emits `{"indices": list[int]}` each tick once `min_points` have arrived:
    /// the indices, within the window, of detected regime changes.
    ///
    /// Args:
    ///     window: Number of recent points in the detection window.
    ///     min_points: Minimum points before detection begins.
    ///     hazard: Prior expected run length between changepoints.
    #[pyo3(signature = (window, min_points=8, hazard=250.0))]
    fn augurs_changepoint(&self, window: usize, min_points: usize, hazard: f64) -> PyStream {
        PyStream(crate::py_augurs::py_augurs_changepoint_inner(
            &self.0, window, min_points, hazard,
        ))
    }

    /// Detect seasonal periods in this stream of floats (periodogram over a
    /// sliding window).
    ///
    /// Emits `{"periods": list[int]}` each tick once enough points have arrived.
    ///
    /// Args:
    ///     window: Number of recent points in the detection window.
    ///     min_points: Minimum points before detection begins, or None.
    ///     min_period: Shortest period to consider, or None.
    ///     max_period: Longest period to consider, or None.
    #[pyo3(signature = (window, min_points=None, min_period=None, max_period=None))]
    fn augurs_seasons(
        &self,
        window: usize,
        min_points: Option<usize>,
        min_period: Option<u32>,
        max_period: Option<u32>,
    ) -> PyStream {
        PyStream(crate::py_augurs::py_augurs_seasons_inner(
            &self.0, window, min_points, min_period, max_period,
        ))
    }

    /// Compute the pairwise dynamic time warping distance matrix over a window
    /// of per-series readings.
    ///
    /// Each value must be a `list[float]` (one reading per series). Emits
    /// `{"rows": list[list[float]]}` each tick, an `n x n` distance matrix.
    ///
    /// Args:
    ///     window: Number of recent samples compared per series.
    ///     metric: "euclidean" (default) or "manhattan".
    #[pyo3(signature = (window, metric="euclidean"))]
    fn augurs_dtw(&self, window: usize, metric: &str) -> PyResult<PyStream> {
        Ok(PyStream(crate::py_augurs::py_augurs_dtw_inner(
            &self.0, window, metric,
        )?))
    }

    /// Cluster the series in a window of per-series readings via DBSCAN over
    /// their pairwise DTW distances.
    ///
    /// Each value must be a `list[float]` (one reading per series). Emits
    /// `{"labels": list[int]}` each tick — a cluster label per series
    /// (`-1` = noise).
    ///
    /// Args:
    ///     window: Number of recent samples compared per series.
    ///     epsilon: DBSCAN neighbourhood radius (max DTW distance).
    ///     min_cluster_size: DBSCAN minimum core-point neighbourhood size.
    ///     metric: "euclidean" (default) or "manhattan".
    #[pyo3(signature = (window, epsilon, min_cluster_size, metric="euclidean"))]
    fn augurs_cluster(
        &self,
        window: usize,
        epsilon: f64,
        min_cluster_size: usize,
        metric: &str,
    ) -> PyResult<PyStream> {
        Ok(PyStream(crate::py_augurs::py_augurs_cluster_inner(
            &self.0,
            window,
            epsilon,
            min_cluster_size,
            metric,
        )?))
    }

    /// Write this stream to a KDB+ table.
    ///
    /// Args:
    ///     host: KDB+ server hostname
    ///     port: KDB+ server port
    ///     table: Name of the target KDB+ table
    ///     columns: List of (name, type) tuples for non-time columns.
    ///              Supported types: "symbol", "float", "long", "int", "bool"
    ///
    /// Returns:
    ///     A Node that drives the write operation.
    #[pyo3(signature = (host, port, table, columns))]
    fn kdb_write(
        &self,
        host: String,
        port: u16,
        table: String,
        columns: Vec<(String, String)>,
    ) -> PyResult<PyNode> {
        let conn = ::wingfoil::adapters::kdb::KdbConnection::new(host, port);
        let node = crate::py_kdb::py_kdb_write_inner(conn, table, columns, &self.0)?;
        Ok(PyNode::new(node))
    }

    /// Write this stream of dicts to a PostgreSQL table.
    ///
    /// The graph timestamp is prepended as the first column, so the target table's
    /// columns must be `(timestamp, <columns in the given order>)`.
    ///
    /// Args:
    ///     conn_str: libpq connection string, e.g.
    ///         `"host=localhost user=postgres password=postgres dbname=postgres"`
    ///     table: target table name (identifier-quoted per dot-separated segment)
    ///     columns: list of (name, type) tuples for the non-time columns. The type
    ///              selects the SQL parameter width and must match the column:
    ///              "bool", "int"/"int4"/"integer" (int4), "long"/"int8"/"bigint"
    ///              (int8), "float4"/"real", "float"/"float8"/"double" (float8),
    ///              "text"/"str", "bytes"/"bytea".
    ///
    /// Stream values must be dicts with the declared column names, or lists of such
    /// dicts for multiple rows per tick. A missing key, an unsupported declared
    /// type, or a wrong-typed value aborts the run — pass an explicit `None` to
    /// write SQL NULL.
    #[pyo3(signature = (conn_str, table, columns))]
    fn postgres_write(
        &self,
        conn_str: String,
        table: String,
        columns: Vec<(String, String)>,
    ) -> PyNode {
        PyNode::new(crate::py_postgres::py_postgres_write_inner(
            &self.0, conn_str, table, columns,
        ))
    }

    /// Publish this stream of dicts to etcd via PUT.
    ///
    /// Stream values must be dicts with `"key"` (str) and `"value"` (bytes),
    /// or lists of such dicts for multiple writes per tick.
    ///
    /// Args:
    ///     endpoint: etcd endpoint, e.g. `"http://localhost:2379"`
    ///     lease_ttl: optional lease TTL in seconds; keys expire after this duration
    ///                and vanish immediately on clean shutdown. Pass `None` for
    ///                persistent keys (default).
    ///     force: if `True` (default), silently overwrite existing keys.
    ///            If `False`, fail if any key already exists.
    ///
    /// Returns:
    ///     A Node that drives the write operation.
    #[cfg(feature = "etcd")]
    #[pyo3(signature = (endpoint, lease_ttl=None, force=true))]
    fn etcd_pub(&self, endpoint: String, lease_ttl: Option<f64>, force: bool) -> PyNode {
        PyNode::new(crate::py_etcd::py_etcd_pub_inner(
            &self.0, endpoint, lease_ttl, force,
        ))
    }

    /// Publish this stream of dicts to a Fluvio topic.
    ///
    /// Stream values must be dicts with `"value"` (bytes) and optional `"key"` (str),
    /// or lists of such dicts for multiple records per tick.
    ///
    /// Args:
    ///     endpoint: Fluvio SC endpoint, e.g. `"127.0.0.1:9003"`
    ///     topic: Fluvio topic name (must already exist)
    ///
    /// Returns:
    ///     A Node that drives the write operation.
    fn fluvio_pub(&self, endpoint: String, topic: String) -> PyNode {
        PyNode::new(crate::py_fluvio::py_fluvio_pub_inner(
            &self.0, endpoint, topic,
        ))
    }

    /// Produce this stream of dicts to Kafka.
    ///
    /// Stream values must be dicts with `"value"` (bytes) and optionally
    /// `"topic"` (str) and `"key"` (bytes), or lists of such dicts for
    /// multi-message writes per tick.
    ///
    /// Args:
    ///     brokers: Kafka bootstrap servers, e.g. `"localhost:9092"`
    ///     topic: Default topic for records that don't specify one
    ///
    /// Returns:
    ///     A Node that drives the produce operation.
    fn kafka_pub(&self, brokers: String, topic: String) -> PyNode {
        PyNode::new(crate::py_kafka::py_kafka_pub_inner(&self.0, brokers, topic))
    }

    /// Publish this stream of dicts to Redis via `PUBLISH`.
    ///
    /// Stream values must be dicts with `"payload"` (bytes) and optionally
    /// `"channel"` (str), or lists of such dicts for multi-message writes per tick.
    ///
    /// Args:
    ///     url: Redis URL, e.g. `"redis://127.0.0.1:6379"`
    ///     channel: default channel for dicts that don't specify one
    ///
    /// Returns:
    ///     A Node that drives the publish operation.
    fn redis_pub(&self, url: String, channel: String) -> PyNode {
        PyNode::new(crate::py_redis::py_redis_pub_inner(&self.0, url, channel))
    }

    /// Append this stream of dicts to a Redis stream via `XADD`.
    ///
    /// Stream values must be dicts with `"fields"` (dict of `name -> bytes`) and
    /// optionally `"key"` (str), or lists of such dicts for multi-entry writes per tick.
    ///
    /// Args:
    ///     url: Redis URL, e.g. `"redis://127.0.0.1:6379"`
    ///     key: default stream key for dicts that don't specify one
    ///
    /// Returns:
    ///     A Node that drives the append operation.
    fn redis_stream_write(&self, url: String, key: String) -> PyNode {
        PyNode::new(crate::py_redis::py_redis_stream_write_inner(
            &self.0, url, key,
        ))
    }

    /// Publish this stream of bytes to a ZMQ PUB socket bound on the given port.
    ///
    /// The stream values must be `bytes` objects. Only supported in real-time mode.
    ///
    /// Args:
    ///     port: TCP port to bind the PUB socket on
    ///
    /// Returns:
    ///     A Node that drives the publish operation.
    fn zmq_pub(&self, port: u16) -> PyNode {
        PyNode::new(crate::py_zmq::py_zmq_pub_inner(&self.0, port))
    }

    // ── Latency stamping ─────────────────────────────────────────────────

    /// Stamp a named latency stage on each tick using the cycle-start
    /// wall-clock time. The stream must carry `TracedBytes` values.
    ///
    /// Args:
    ///     stage: Stage name (must match one of the names in the `Latency`).
    ///
    /// Returns:
    ///     A new Stream with the stage stamped.
    fn stamp(&self, stage: String) -> PyStream {
        PyStream(crate::py_latency::py_stamp_inner(&self.0, stage, false))
    }

    /// Like `stamp`, but only inserts the stamp node when `enabled` is True.
    /// When False, returns the stream unchanged — zero runtime cost.
    #[pyo3(signature = (stage, enabled))]
    fn stamp_if(&self, stage: String, enabled: bool) -> PyStream {
        if enabled {
            self.stamp(stage)
        } else {
            self.clone()
        }
    }

    /// Stamp a named latency stage with a precise wall-clock read (~5-10ns
    /// TSC read per tick). Gives intra-cycle resolution.
    fn stamp_precise(&self, stage: String) -> PyStream {
        PyStream(crate::py_latency::py_stamp_inner(&self.0, stage, true))
    }

    /// Like `stamp_precise`, but only inserts the stamp node when `enabled`
    /// is True. When False, returns the stream unchanged.
    #[pyo3(signature = (stage, enabled))]
    fn stamp_precise_if(&self, stage: String, enabled: bool) -> PyStream {
        if enabled {
            self.stamp_precise(stage)
        } else {
            self.clone()
        }
    }

    /// Install a latency report sink. The stream must carry `TracedBytes`
    /// values. Per-stage delta statistics (count/min/mean/p50/p99/max) are
    /// printed on graph shutdown.
    ///
    /// Args:
    ///     stages: Stage names in order (same list used for `Latency`).
    ///     print_on_teardown: Whether to print the report on shutdown (default True).
    ///
    /// Returns:
    ///     A Node that drives the report sink.
    #[pyo3(signature = (stages, print_on_teardown=true))]
    fn latency_report(&self, stages: Vec<String>, print_on_teardown: bool) -> PyNode {
        PyNode::new(crate::py_latency::py_latency_report_inner(
            &self.0,
            stages,
            print_on_teardown,
        ))
    }

    /// Like `latency_report`, but only installs the sink when `enabled` is
    /// True. When False, returns the upstream as a Node (no report sink).
    #[pyo3(signature = (stages, enabled, print_on_teardown=true))]
    fn latency_report_if(
        &self,
        stages: Vec<String>,
        enabled: bool,
        print_on_teardown: bool,
    ) -> PyNode {
        if enabled {
            self.latency_report(stages, print_on_teardown)
        } else {
            PyNode::new(self.0.clone().as_node())
        }
    }

    /// Publish this stream to a [`WebServer`] topic over WebSocket.
    ///
    /// Values must be JSON-compatible Python objects (dict / list /
    /// str / int / float / bool / bytes / None). Connected browser
    /// clients that subscribed to `topic` receive the frames.
    ///
    /// Args:
    ///     server: A WebServer instance.
    ///     topic: The topic name to publish on.
    ///
    /// Returns:
    ///     A Node that drives the publish operation.
    fn web_pub(&self, server: &crate::py_web::PyWebServer, topic: String) -> PyNode {
        PyNode::new(crate::py_web::py_web_pub_inner(
            &self.0,
            server.inner_ref(),
            topic,
        ))
    }

    /// Publish this stream of bytes to an iceoryx2 service.
    ///
    /// When `stages` is provided, expects the stream to carry `TracedBytes`
    /// values and serializes as `latency_header + payload_bytes` on the wire.
    ///
    /// Args:
    ///     service_name: iceoryx2 service name, e.g. `"my/service"`
    ///     variant: Service variant ("ipc" or "local")
    ///     history_size: Service history ring size (must match subscribers)
    ///     initial_max_slice_len: Initial maximum slice length (bytes)
    ///     stages: Optional list of latency stage names for instrumented mode
    ///
    /// Returns:
    ///     A Node that drives the publish operation.
    #[cfg(feature = "iceoryx2")]
    #[pyo3(signature = (service_name, variant=crate::py_iceoryx2::PyIceoryx2ServiceVariant::Ipc, history_size=5, initial_max_slice_len=128*1024, stages=None))]
    fn iceoryx2_pub(
        &self,
        service_name: String,
        variant: crate::py_iceoryx2::PyIceoryx2ServiceVariant,
        history_size: usize,
        initial_max_slice_len: usize,
        stages: Option<Vec<String>>,
    ) -> PyNode {
        PyNode::new(crate::py_iceoryx2::py_iceoryx2_pub_inner(
            &self.0,
            service_name,
            variant,
            history_size,
            initial_max_slice_len,
            stages,
        ))
    }

    /// Publish this stream of `bytes` to an Aeron channel.
    ///
    /// Each stream value is published as one Aeron message. Requires a running
    /// media driver — connection happens at call time and raises
    /// `RuntimeError` if the driver is unreachable or the publication cannot be
    /// resolved within `timeout_secs`.
    ///
    /// Args:
    ///     channel: Aeron channel URI, e.g. `"aeron:ipc"`
    ///     stream_id: Aeron stream id
    ///     timeout_secs: How long to wait for the publication to resolve
    ///
    /// Returns:
    ///     A Node that drives the publish operation.
    #[cfg(feature = "aeron")]
    #[pyo3(signature = (channel, stream_id, timeout_secs=5.0))]
    fn aeron_pub(&self, channel: String, stream_id: i32, timeout_secs: f64) -> PyResult<PyNode> {
        Ok(PyNode::new(crate::py_aeron::py_aeron_pub_inner(
            &self.0,
            channel,
            stream_id,
            timeout_secs,
        )?))
    }

    /// Push this stream as an OTLP gauge metric.
    ///
    /// Args:
    ///     metric_name: Name of the metric to report
    ///     endpoint: OTLP HTTP endpoint, e.g. `"http://localhost:4318"`
    ///     service_name: Service name reported in OTLP resource attributes
    ///
    /// Returns:
    ///     A Node that drives the push operation.
    fn otlp_push(&self, metric_name: String, endpoint: String, service_name: String) -> PyNode {
        crate::py_otlp::py_otlp_push_inner(self, metric_name, endpoint, service_name)
    }

    /// Publish this stream of bytes and register as `name` in etcd.
    ///
    /// Binds on `127.0.0.1`; use `zmq_pub_etcd_on` for multi-host deployments
    /// where `127.0.0.1` is not routable by subscribers on other hosts.
    ///
    /// Args:
    ///     name: Name / etcd key to register under (e.g. "quotes")
    ///     port: TCP port to bind the PUB socket on
    ///     endpoint: etcd endpoint (e.g. "http://localhost:2379")
    ///
    /// Returns:
    ///     A Node that drives the publish operation.
    #[cfg(feature = "etcd")]
    fn zmq_pub_etcd(&self, name: String, port: u16, endpoint: String) -> PyNode {
        PyNode::new(crate::py_zmq::py_zmq_pub_etcd_inner(
            &self.0, name, port, endpoint,
        ))
    }

    /// Like `zmq_pub_etcd` but binds on `address` instead of `127.0.0.1`.
    ///
    /// Args:
    ///     name: Name / etcd key to register under
    ///     address: Routable bind address (e.g. "192.168.1.10")
    ///     port: TCP port to bind the PUB socket on
    ///     endpoint: etcd endpoint
    ///
    /// Returns:
    ///     A Node that drives the publish operation.
    #[cfg(feature = "etcd")]
    fn zmq_pub_etcd_on(
        &self,
        name: String,
        address: String,
        port: u16,
        endpoint: String,
    ) -> PyNode {
        PyNode::new(crate::py_zmq::py_zmq_pub_etcd_on_inner(
            &self.0, name, address, port, endpoint,
        ))
    }

    // end StreamOperators
}