Merge pull request #11 from CU-ESIIL/codex/add-pipe-system-and-ops-layout

Add pipe helper and ops verbs

Author: ttuff

README.md

@@ -52,6 +52,21 @@ var_cube = cd.variance_cube(cube)
 var_cube.to_netcdf("gridmet_variance.nc")
 ```
 
+Pipe ggplot-style operations with ``|`` for quick cube math:
+
+```python
+import cubedynamics as cd
+
+cube = ...  # any xarray DataArray or Dataset
+
+result = (
+    cd.pipe(cube)
+    | cd.anomaly(dim="time")
+    | cd.month_filter([6, 7, 8])
+    | cd.variance(dim="time")
+).unwrap()
+```
+
 Additional helpers can build NDVI z-score cubes, compute rolling correlation vs
 an anchor pixel, or export “lexcubes” for downstream dashboards. Follow the docs
 for more end-to-end examples while the streaming implementations are finalized.

docs/getting_started.md

@@ -59,3 +59,38 @@ print(cube)
 
 If the call succeeds you are ready to work through the [Concepts](concepts.md)
 and [API & Examples](climate_cubes.md) sections.
+
+## Pipe syntax (ggplot-style verbs)
+
+CubeDynamics now exposes a ``pipe()`` helper and pipeable verbs under
+``cubedynamics.ops``. Wrap any xarray ``DataArray`` or ``Dataset`` with
+``cd.pipe()`` and compose operations with the ``|`` operator:
+
+```python
+import cubedynamics as cd
+
+cube = ...
+
+result = (
+    cd.pipe(cube)
+    | cd.anomaly(dim="time")
+    | cd.month_filter([6, 7, 8])
+    | cd.variance(dim="time")
+    | cd.to_netcdf("out.nc")
+).unwrap()
+```
+
+Pipeable verbs are factories. You can create your own by following the same
+pattern:
+
+```python
+def my_custom_op(scale):
+    def _inner(cube):
+        return cube * scale
+    return _inner
+
+result = cd.pipe(cube) | my_custom_op(0.5)
+```
+
+This keeps the streaming-first philosophy: each verb simply transforms or
+reduces the cube provided by the pipe without forcing eager downloads.

src/cubedynamics/__init__.py

@@ -8,6 +8,7 @@
 """
 
 from .version import __version__
+from .piping import Pipe, pipe
 
 # Legacy, fully implemented APIs -------------------------------------------------
 from .data.gridmet import load_gridmet_cube
@@ -30,7 +31,8 @@
 # Streaming-first stubs for the new architecture ---------------------------------
 from .gridmet_streaming import stream_gridmet_to_cube
 from .prism_streaming import stream_prism_to_cube
-from .correlation_cubes import correlation_cube
+from .correlation_cubes import correlation_cube as streaming_correlation_cube
+from .ops import anomaly, month_filter, variance, correlation_cube, to_netcdf
 
 __all__ = [
     "__version__",
@@ -54,5 +56,12 @@
     # Streaming-first stubs -------------------------------------------------------
     "stream_gridmet_to_cube",
     "stream_prism_to_cube",
+    "Pipe",
+    "pipe",
+    "anomaly",
+    "month_filter",
+    "variance",
     "correlation_cube",
+    "to_netcdf",
+    "streaming_correlation_cube",
 ]

src/cubedynamics/ops/__init__.py

@@ -0,0 +1,13 @@
+"""Pipeable cube operations."""
+
+from .transforms import anomaly, month_filter
+from .stats import variance, correlation_cube
+from .io import to_netcdf
+
+__all__ = [
+    "anomaly",
+    "month_filter",
+    "variance",
+    "correlation_cube",
+    "to_netcdf",
+]

src/cubedynamics/ops/io.py

@@ -0,0 +1,18 @@
+"""I/O helpers for pipe chains."""
+
+from __future__ import annotations
+
+import xarray as xr
+
+
+def to_netcdf(path: str, **to_netcdf_kwargs):
+    """Factory for a pipeable ``.to_netcdf`` side-effect operation."""
+
+    def _inner(da: xr.DataArray | xr.Dataset):
+        da.to_netcdf(path, **to_netcdf_kwargs)
+        return da
+
+    return _inner
+
+
+__all__ = ["to_netcdf"]

src/cubedynamics/ops/stats.py

@@ -0,0 +1,39 @@
+"""Statistical pipeable operations."""
+
+from __future__ import annotations
+
+import xarray as xr
+
+
+def variance(dim: str = "time"):
+    """Factory returning a variance reducer over ``dim`` for pipe chains."""
+
+    def _inner(da: xr.DataArray | xr.Dataset):
+        if dim not in da.dims:
+            raise ValueError(f"Dimension {dim!r} not found in object dims: {da.dims}")
+        return da.var(dim=dim)
+
+    return _inner
+
+
+def correlation_cube(other: xr.DataArray | xr.Dataset | None, dim: str = "time"):
+    """Factory placeholder for a future correlation cube operation.
+
+    Parameters
+    ----------
+    other:
+        The comparison cube captured by the factory.
+    dim:
+        Dimension over which correlations would be computed once implemented.
+    """
+
+    if other is None or not isinstance(dim, str):
+        raise NotImplementedError("correlation_cube is not implemented yet.")
+
+    def _inner(da: xr.DataArray | xr.Dataset):  # pragma: no cover - stub
+        raise NotImplementedError("correlation_cube is not implemented yet.")
+
+    return _inner
+
+
+__all__ = ["variance", "correlation_cube"]

src/cubedynamics/ops/transforms.py

@@ -0,0 +1,52 @@
+"""Transform-style pipeable operations."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+import xarray as xr
+
+
+def anomaly(dim: str = "time"):
+    """Factory for an anomaly transform over ``dim``.
+
+    Returns a callable that subtracts the mean over ``dim`` while preserving the
+    input type (``DataArray`` or ``Dataset``). The returned callable is intended
+    to be used within a :class:`~cubedynamics.piping.Pipe` chain via ``|``.
+    """
+
+    def _inner(da: xr.DataArray | xr.Dataset):
+        if dim not in da.dims:
+            raise ValueError(f"Dimension {dim!r} not found in object dims: {da.dims}")
+        return da - da.mean(dim=dim)
+
+    return _inner
+
+
+def month_filter(months: Iterable[int]):
+    """Factory for filtering calendar months from a ``time`` coordinate.
+
+    Parameters
+    ----------
+    months:
+        Sequence of integers (1-12) to keep. Requires that the object has a
+        ``time`` coordinate with datetime-like values.
+    """
+
+    months = tuple(int(m) for m in months)
+
+    def _inner(da: xr.DataArray | xr.Dataset):
+        if "time" not in da.coords:
+            raise ValueError("month_filter requires a 'time' coordinate.")
+        time = da["time"]
+        try:
+            month_vals = time.dt.month
+        except Exception as exc:  # pragma: no cover - dt errors raised as ValueError below
+            raise ValueError("month_filter: 'time' coordinate must be datetime-like.") from exc
+        mask = month_vals.isin(months)
+        return da.where(mask, drop=True)
+
+    return _inner
+
+
+__all__ = ["anomaly", "month_filter"]

src/cubedynamics/piping.py

@@ -0,0 +1,50 @@
+"""Pipe wrapper enabling ``|`` composition for cube operations."""
+
+from __future__ import annotations
+
+from typing import Callable, Generic, TypeVar
+
+
+T = TypeVar("T")
+
+
+U = TypeVar("U")
+
+
+class Pipe(Generic[T]):
+    """Wrap a value so it can flow through ``|`` pipe stages."""
+
+    def __init__(self, value: T) -> None:
+        self.value = value
+
+    def __or__(self, func: Callable[[T], U]) -> "Pipe[U]":
+        """Apply ``func`` to the wrapped value and return a new :class:`Pipe`."""
+
+        new_value = func(self.value)
+        return Pipe(new_value)
+
+    def unwrap(self) -> T:
+        """Return the wrapped value, ending the pipe chain."""
+
+        return self.value
+
+    @property
+    def v(self) -> T:
+        """Convenience alias for :pyattr:`value`."""
+
+        return self.value
+
+
+def pipe(value: T) -> Pipe[T]:
+    """Wrap ``value`` in a :class:`Pipe`, enabling ``Pipe | op(...)`` syntax.
+
+    Example
+    -------
+    >>> result = (pipe(1) | (lambda x: x + 2)).unwrap()
+    >>> assert result == 3
+    """
+
+    return Pipe(value)
+
+
+__all__ = ["Pipe", "pipe"]

src/cubedynamics/tests/test_piping_ops.py

@@ -0,0 +1,49 @@
+"""Tests for the ggplot-style pipe syntax and verbs."""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+import xarray as xr
+
+import cubedynamics as cd
+
+
+def _make_time_series(count: int = 12):
+    time = pd.date_range("2000-01-01", periods=count, freq="MS")
+    data = xr.DataArray(
+        np.arange(count, dtype=float),
+        dims=("time",),
+        coords={"time": time},
+    )
+    return data
+
+
+def test_pipe_basic_chain():
+    da = _make_time_series()
+
+    result = (cd.pipe(da) | cd.anomaly(dim="time") | cd.variance(dim="time")).unwrap()
+
+    assert isinstance(result, xr.DataArray)
+    assert result.dims == ()
+    assert float(result) >= 0
+
+
+def test_month_filter_reduces_time():
+    da = _make_time_series(24)
+
+    summer = (cd.pipe(da) | cd.month_filter([6, 7, 8])).unwrap()
+
+    assert set(int(m) for m in summer["time"].dt.month.values) == {6, 7, 8}
+
+
+def test_to_netcdf_roundtrip(tmp_path):
+    da = _make_time_series()
+    path = tmp_path / "out.nc"
+
+    result = (cd.pipe(da) | cd.to_netcdf(path)).unwrap()
+
+    assert path.exists()
+    loaded = xr.load_dataarray(path)
+    xr.testing.assert_identical(da, loaded)
+    xr.testing.assert_identical(da, result)