Composition¶

The composition primitives are carrier-agnostic and live in pipekit; the xarray-aware Operator base and the Dataset combinators are xrops:

from pipekit import Sequential, Graph, Input, Node
from xrops import Augment, Operator

xrops.Operator is a thin xarray-aware subclass of pipekit.Operator that adds xarray.DataTree dispatch to __call__ — every operator in the library inherits it, so all of them map leaf-wise over a DataTree for free. The xarray-Dataset-specific combinators (Augment, ApplyToEach) live in xrops.combinators.

Linear and DAG composition, same operators

from pipekit import Sequential, Graph, Input
from xrops.anomaly import RemoveMean
from xrops.ocn import Streamfunction, GeostrophicVelocities

# Linear pipeline
pipe = Sequential(RemoveMean(dim=("lat", "lon")), GeostrophicVelocities())
out = pipe(ds)

# The same operators wired as a DAG
x = Input("ssh")
psi = Streamfunction()(x)
uv = GeostrophicVelocities()(psi)
graph = Graph(inputs=x, outputs=uv)

Operator base class¶

Operator ¶

Bases: Operator

pipekit.Operator plus xarray DataTree dispatch.

Subclasses implement _apply(carrier, *extra) exactly as they would against pipekit.Operator — the override only affects __call__ dispatch, not _apply.

Example

One _apply implementation serves Dataset and DataTree callers alike — the tree case maps over every leaf:

>>> import numpy as np
>>> import xarray as xr
>>> from xrops import Operator
>>> class Scale(Operator):
...     def __init__(self, factor: float) -> None:
...         self.factor = factor
...     def _apply(self, ds: xr.Dataset) -> xr.Dataset:
...         return ds * self.factor
>>> ds = xr.Dataset({"ssh": ("x", np.arange(3.0))})
>>> Scale(10.0)(ds)["ssh"].values
array([ 0., 10., 20.])
>>> tree = xr.DataTree.from_dict({"coarse": ds, "fine": ds * 2})
>>> out = Scale(10.0)(tree)
>>> sorted(out.children), float(out["fine"]["ssh"][2])
(['coarse', 'fine'], 40.0)

Sequential pipelines¶

Sequential ¶

Sequential(operators: list[Operator])

Bases: Operator

Apply a list of operators in order, threading the output of each into the next.

Parameters:

Name	Type	Description	Default
`operators`	`list[Operator]`	A list of `Operator` instances. The empty list is allowed but degenerate: `Sequential([])(x)` returns `x` unchanged, while `Sequential([])()` raises `TypeError` (the empty pipeline has no first operator to consult, and the carrier has to come from somewhere).	required

Raises:

Type	Description
`TypeError`	if any element is not an `Operator`, or if any non-final element is marked `_terminal`.

Example

Basic::

pipe = Sequential([Scale(0.5), Scale(2.0)])
assert pipe(10.0) == 10.0

Slice into a sub-pipeline::

head = pipe[:1]   # Sequential([Scale(0.5)])

Index a single step::

first = pipe[0]   # Scale(0.5)

Source code in .venv/lib/python3.12/site-packages/pipekit/_base/sequential.py

def __init__(self, operators: list[Operator]) -> None:
    for i, op in enumerate(operators):
        require_operator(op, "Sequential", f"operators[{i}]")
    for i, op in enumerate(operators[:-1]):
        if op._terminal:
            raise TypeError(
                f"Sequential[{i}] is a terminal operator "
                f"({type(op).__name__}) — terminal operators are only "
                "valid as the last step of a Sequential."
            )
    self.operators = list(operators)

compute_output_signature ¶

compute_output_signature(input_signature: Any) -> Any

Thread the signature through every operator in order.

A registered carrier is read as its Signature first.

Source code in .venv/lib/python3.12/site-packages/pipekit/_base/sequential.py

def compute_output_signature(self, input_signature: Any) -> Any:
    """Thread the signature through every operator in order.

    A registered carrier is read as its `Signature` first.
    """
    sig = signature_of(input_signature)
    for op in self.operators:
        sig = op.compute_output_signature(sig)
        if sig is None:
            return None
    return sig

describe ¶

describe(indent: int = 0) -> str

Return an indented-tree pretty-print of the pipeline.

Recurses into nested Sequential / Graph operators. Useful for notebook exploration; not parsed.

Source code in .venv/lib/python3.12/site-packages/pipekit/_base/sequential.py

def describe(self, indent: int = 0) -> str:
    """Return an indented-tree pretty-print of the pipeline.

    Recurses into nested `Sequential` / `Graph` operators. Useful
    for notebook exploration; not parsed.
    """
    pad = "  " * indent
    if not self.operators:
        return f"{pad}Sequential([])"
    lines: list[str] = [f"{pad}Sequential(["]
    for op in self.operators:
        if isinstance(op, Sequential):
            lines.append(op.describe(indent=indent + 1) + ",")
        else:
            lines.append(f"{pad}  {op!r},")
    lines.append(f"{pad}])")
    return "\n".join(lines)

summary ¶

summary(input_signature: Any = None) -> str

Render a Keras-style summary table.

Threads input_signature through compute_output_signature; a registered carrier is read as its Signature first. If any operator returns None (does not track shape), the summary still lists the operator but shows ? for its output signature. With input_signature=None, only operator names and types are shown.

The return type is str rather than printing directly so the caller decides where it goes (notebook, log, file).

Source code in .venv/lib/python3.12/site-packages/pipekit/_base/sequential.py

def summary(self, input_signature: Any = None) -> str:
    """Render a Keras-style summary table.

    Threads ``input_signature`` through `compute_output_signature`;
    a registered carrier is read as its `Signature` first. If any
    operator returns ``None`` (does not track shape), the summary
    still lists the operator but shows ``?`` for its output
    signature. With ``input_signature=None``, only operator names
    and types are shown.

    The return type is ``str`` rather than printing directly so the
    caller decides where it goes (notebook, log, file).
    """
    rows: list[tuple[str, str, str]] = []
    sig = signature_of(input_signature)
    for i, op in enumerate(self.operators):
        sig_in = "?" if sig is None else repr(sig)
        sig = op.compute_output_signature(sig)
        sig_out = "?" if sig is None else repr(sig)
        rows.append((f"[{i}] {type(op).__name__}", sig_in, sig_out))

    # Column widths
    w_name = max((len(r[0]) for r in rows), default=4)
    w_in = max((len(r[1]) for r in rows), default=8)
    w_out = max((len(r[2]) for r in rows), default=8)

    header = (
        f"{'Operator'.ljust(w_name)}  "
        f"{'Input sig'.ljust(w_in)}  "
        f"{'Output sig'.ljust(w_out)}"
    )
    rule = "-" * len(header)
    body = [
        f"{name.ljust(w_name)}  {sin.ljust(w_in)}  {sout.ljust(w_out)}"
        for name, sin, sout in rows
    ]
    return "\n".join([header, rule, *body])

Functional Graph API¶

Input ¶

Input(name: str)

Bases: Node

A named entry point into a Graph.

Operator.__call__ recognises Node (and any subclass, including Input) as graph-construction mode. Graph._apply supplies the value at run-time from the caller's kwargs / positional args.

Subclassing Node (master plan Report 2.A.3, "xr_toolz shape") lets the topological sort treat all vertices uniformly with one special case: "this node has no operator — source it from the caller's inputs".

Source code in .venv/lib/python3.12/site-packages/pipekit/_base/graph.py

def __init__(self, name: str) -> None:
    super().__init__(operator=None, parents=())
    self.name = name

Node ¶

Node(operator: Operator | None, parents: tuple[Any, ...])

A vertex in a Graph.

Created automatically by Operator.__call__ when any positional argument is a Node (or Input, which subclasses Node). Carries the operator and its parent vertices. Compared by identity so the id(...)-keyed evaluation cache during topological sort works.

Raises:

Type	Description
`TypeError`	if any element of `parents` is not a `Node`. `Graph` evaluation walks `parents` recursively and indexes an `id(...)`-keyed cache; non-Node parents would crash with an unhelpful `KeyError` deep in the traversal.

Source code in .venv/lib/python3.12/site-packages/pipekit/_base/graph.py

def __init__(
    self,
    operator: Operator | None,
    parents: tuple[Any, ...],
) -> None:
    for i, p in enumerate(parents):
        if not isinstance(p, Node):
            raise TypeError(
                f"Node.parents[{i}] is {type(p).__name__}, expected Node. "
                "Wrap literal values in `Const(value)` to lift them into "
                "the graph."
            )
    self.operator = operator
    self.parents = parents

Graph ¶

Graph(inputs: dict[str, Input], outputs: dict[str, Node])

Bases: Operator

A symbolic operator graph with multiple inputs and outputs.

Construction is implicit — calling operators on Input / Node instances builds the graph; Graph(inputs=..., outputs=...) wraps the result. _apply(**inputs) evaluates in topological order.

Inherits from Operator so a Graph satisfies the same interface as any other operator. Operator.__call__ dispatches keyword args straight through to Graph._apply; positional args are bound to declared Inputs in declaration order.

Parameters:

Name	Type	Description	Default
`inputs`	`dict[str, Input]`	Map of `input-name → Input` placeholders. The keys are the keyword names accepted by `__call__`.	required
`outputs`	`dict[str, Node]`	Map of `output-name → Node` (or `Input`, if the output is a direct passthrough). The keys are the keys of the returned dict.	required

State

A graph that contains a stateful node (a StatefulOperator, or a Sequential / Graph that contains one) threads a carry state the way Sequential does: call it as graph(a=..., b=..., state=s) and it returns (outputs, state). The state passes through the stateful nodes in topological order (the order describe lists); stateless nodes ignore it. Inside a stateful Sequential the graph receives (carrier, state) positionally, like any stateful step.

Raises:

Type	Description
`ValueError`	if the graph contains a cycle, if an `Input` referenced by a node isn't declared in `inputs`, if a required input is missing at call time, or if a stateful graph declares an input named `state` (the name is the carry state's).

Source code in .venv/lib/python3.12/site-packages/pipekit/_base/graph.py

def __init__(
    self,
    inputs: dict[str, Input],
    outputs: dict[str, Node],
) -> None:
    self.inputs = inputs
    self.outputs = outputs
    self._order = self._topological_sort()
    if self._is_stateful and "state" in inputs:
        raise ValueError(
            "A stateful Graph cannot declare an input named 'state': "
            "`state=` is how its carry state is passed. Rename the input."
        )

get_config ¶

get_config() -> dict[str, Any]

The whole topology: inputs, operators, nodes and outputs.

Every node is recorded, upstream ones included, so two graphs that differ anywhere have different configs (and so different state records and content hashes). The record::

{
    "inputs": ["image", "reference"],
    "operators": [<state>, ...],
    "nodes": [{"operator": 0, "parents": [{"input": "image"}]}, ...],
    "outputs": {"ndvi": {"node": 0}, "raw": {"input": "image"}},
}

operators lists each distinct operator once, by identity, in first-use order, so an operator shared by several nodes is not duplicated. nodes is in topological order; a parent or an output refers either to a declared input by name or to an earlier node by index.

Source code in .venv/lib/python3.12/site-packages/pipekit/_base/graph.py

def get_config(self) -> dict[str, Any]:
    """The whole topology: inputs, operators, nodes and outputs.

    Every node is recorded, upstream ones included, so two graphs
    that differ anywhere have different configs (and so different
    `state` records and content hashes). The record::

        {
            "inputs": ["image", "reference"],
            "operators": [<state>, ...],
            "nodes": [{"operator": 0, "parents": [{"input": "image"}]}, ...],
            "outputs": {"ndvi": {"node": 0}, "raw": {"input": "image"}},
        }

    ``operators`` lists each distinct operator once, by identity, in
    first-use order, so an operator shared by several nodes is not
    duplicated. ``nodes`` is in topological order; a parent or an
    output refers either to a declared input by name or to an
    earlier node by index.
    """
    input_names = {id(node): name for name, node in self.inputs.items()}
    node_index = {id(node): i for i, node in enumerate(self._order)}
    operator_index: dict[int, int] = {}
    operators: list[dict[str, Any]] = []
    nodes: list[dict[str, Any]] = []

    def ref(node: Node) -> dict[str, Any]:
        if id(node) in input_names:
            return {"input": input_names[id(node)]}
        return {"node": node_index[id(node)]}

    for node in self._order:
        assert node.operator is not None  # guaranteed by _topological_sort
        op_id = id(node.operator)
        if op_id not in operator_index:
            operator_index[op_id] = len(operators)
            operators.append(node.operator.state)
        parents = [ref(parent) for parent in node.parents]
        nodes.append({"operator": operator_index[op_id], "parents": parents})

    return {
        "inputs": list(self.inputs),
        "operators": operators,
        "nodes": nodes,
        "outputs": {name: ref(node) for name, node in self.outputs.items()},
    }

from_config `classmethod` ¶

from_config(config: dict[str, Any]) -> Self

Rebuild a graph from the topology get_config records.

Re-creates each Input, then calls each node's operator on its parents in the recorded order, which wires the nodes exactly as they were; an operator shared by several nodes stays one object.

Parameters:

Name	Type	Description	Default
`config`	`dict[str, Any]`	A `get_config` record whose `operators` have already been rebuilt into `Operator` instances.	required

Raises:

Type	Description
`ValueError`	if a reference names an undeclared input or a node that comes later in the order.

Source code in .venv/lib/python3.12/site-packages/pipekit/_base/graph.py

@classmethod
def from_config(cls, config: dict[str, Any]) -> Self:
    """Rebuild a graph from the topology `get_config` records.

    Re-creates each `Input`, then calls each node's operator on its
    parents in the recorded order, which wires the nodes exactly as
    they were; an operator shared by several nodes stays one object.

    Args:
        config: A `get_config` record whose ``operators`` have already
            been rebuilt into `Operator` instances.

    Raises:
        ValueError: if a reference names an undeclared input or a node
            that comes later in the order.
    """
    inputs = {name: Input(name) for name in config["inputs"]}
    operators: list[Operator] = config["operators"]
    built: list[Node] = []

    def lookup(ref: dict[str, Any]) -> Node:
        if "input" in ref:
            if ref["input"] not in inputs:
                raise ValueError(f"Graph config: undeclared input {ref!r}.")
            return inputs[ref["input"]]
        index = ref["node"]
        if not 0 <= index < len(built):
            raise ValueError(f"Graph config: {ref!r} is not an earlier node.")
        return built[index]

    for spec in config["nodes"]:
        parents = [lookup(ref) for ref in spec["parents"]]
        built.append(operators[spec["operator"]](*parents))
    outputs = {name: lookup(ref) for name, ref in config["outputs"].items()}
    return cls(inputs=inputs, outputs=outputs)

compute_output_signature ¶

compute_output_signature(input_signature: Any) -> Any

None: a graph returns a dict of outputs, not one carrier.

Use output_signatures for each output's signature, and summary for the whole table.

Source code in .venv/lib/python3.12/site-packages/pipekit/_base/graph.py

def compute_output_signature(self, input_signature: Any) -> Any:
    """``None``: a graph returns a dict of outputs, not one carrier.

    Use `output_signatures` for each output's signature, and
    `summary` for the whole table.
    """
    return None

output_signatures ¶

output_signatures(input_signatures: Any) -> dict[str, Any]

Each output's signature, propagated node by node without data.

Parameters:

Name	Type	Description	Default
`input_signatures`	`Any`	`{input name: Signature}`; a missing name means unknown. A single-input graph also takes the bare signature.	required

Returns:

Type	Description
`dict[str, Any]`	`{output name: Signature \| None}`; `None` where an
`dict[str, Any]`	operator does not track shape, or where a multi-input node's
`dict[str, Any]`	inputs disagree (see `summary`).

Source code in .venv/lib/python3.12/site-packages/pipekit/_base/graph.py

def output_signatures(self, input_signatures: Any) -> dict[str, Any]:
    """Each output's signature, propagated node by node without data.

    Args:
        input_signatures: ``{input name: Signature}``; a missing name
            means unknown. A single-input graph also takes the bare
            signature.

    Returns:
        ``{output name: Signature | None}``; ``None`` where an
        operator does not track shape, or where a multi-input node's
        inputs disagree (see `summary`).
    """
    signatures = self._node_signatures(input_signatures)
    return {name: signatures[id(node)] for name, node in self.outputs.items()}

summary ¶

summary(input_signatures: Any = None) -> str

A Keras-style table of the nodes, their input and output signatures.

Nodes are listed in topological order. Signatures propagate through each node's compute_output_signature; a node with several inputs gets its operator's signature for them only when they all agree (an elementwise join), and ? otherwise, since compute_output_signature takes one input. The outputs follow.

Parameters:

Name	Type	Description	Default
`input_signatures`	`Any`	As for `output_signatures`; `None` lists the structure with every signature unknown.	`None`

Source code in .venv/lib/python3.12/site-packages/pipekit/_base/graph.py

def summary(self, input_signatures: Any = None) -> str:
    """A Keras-style table of the nodes, their input and output signatures.

    Nodes are listed in topological order. Signatures propagate
    through each node's `compute_output_signature`; a node with
    several inputs gets its operator's signature for them only when
    they all agree (an elementwise join), and ``?`` otherwise, since
    `compute_output_signature` takes one input. The outputs follow.

    Args:
        input_signatures: As for `output_signatures`; ``None`` lists
            the structure with every signature unknown.
    """
    signatures = self._node_signatures(input_signatures)
    names = {id(node): name for name, node in self.inputs.items()}

    def label(node: Node) -> str:
        if id(node) in names:
            return names[id(node)]
        return f"[{self._order.index(node)}]"

    def show(signature: Any) -> str:
        if signature is None:
            return "?"
        # a Signature's compact "(y=64, x=64) float32" form
        fmt = getattr(signature, "format", None)
        return fmt() if callable(fmt) else repr(signature)

    rows: list[tuple[str, str, str]] = []
    for i, node in enumerate(self._order):
        parents = ", ".join(label(parent) for parent in node.parents)
        sig_in = ", ".join(show(signatures[id(p)]) for p in node.parents) or "-"
        name = f"[{i}] {type(node.operator).__name__}({parents})"
        rows.append((name, sig_in, show(signatures[id(node)])))
    for out_name, node in self.outputs.items():
        rows.append(
            (f"→ {out_name} = {label(node)}", "", show(signatures[id(node)]))
        )

    w_name = max([len("Node"), *(len(r[0]) for r in rows)])
    w_in = max([len("Input sig"), *(len(r[1]) for r in rows)])
    header = f"{'Node'.ljust(w_name)}  {'Input sig'.ljust(w_in)}  Output sig"
    body = [f"{n.ljust(w_name)}  {i.ljust(w_in)}  {o}" for n, i, o in rows]
    return "\n".join([header, "-" * len(header), *body])

describe ¶

describe() -> str

Return a short topology description.

Lists inputs, internal nodes (in topological order), and outputs. Useful for notebook exploration.

Source code in .venv/lib/python3.12/site-packages/pipekit/_base/graph.py

def describe(self) -> str:
    """Return a short topology description.

    Lists inputs, internal nodes (in topological order), and
    outputs. Useful for notebook exploration.
    """
    lines = [
        f"Graph(inputs={list(self.inputs)}, outputs={list(self.outputs)})",
        "  internal nodes (topological order):",
    ]
    for i, node in enumerate(self._order):
        op_name = type(node.operator).__name__ if node.operator else "Input"
        lines.append(f"    [{i}] {op_name}")
    return "\n".join(lines)

Observation taps¶

Tap is a generic pass-through observer from pipekit — it runs a side-effecting callback (logging, plotting, assertions) without altering the data flowing through.

Tap ¶

Tap(fn: Callable[[Any], Any], name: str | None = None)

Bases: Operator

Fire a callback on the carrier; return the carrier unchanged.

Distinct from Sink only by naming convention — Tap is for observation (logging, metrics, breakpoint hooks); Sink is for persistence. Both pass the carrier through; both carry forbid_in_yaml = True.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[[Any], Any]`	Callable invoked with the carrier. Its return value is ignored.	required
`name`	`str \| None`	Optional name for `__repr__`. Defaults to the callable's `__name__` if available, otherwise `"Tap"`.	`None`

Example::

log = Tap(lambda x: print(getattr(x, "shape", x)), name="log_shape")
pipe = Scale(2.0) | log | AddConst(1.0)

Source code in .venv/lib/python3.12/site-packages/pipekit/observe.py

def __init__(self, fn: Callable[[Any], Any], name: str | None = None) -> None:
    self.fn = fn
    self.name = name or getattr(fn, "__name__", "Tap")

xarray-Dataset combinators¶

Augment merges an operator's output back into its input Dataset (rather than replacing it); ApplyToEach maps an operator over every data variable independently.

Augment ¶

Augment(inner: Operator)

Bases: Operator

Run an inner Operator and merge its output back into the input.

The inner operator must accept a Dataset and return a Dataset. The return value of :class:Augment is input.merge(inner(input)) — the input's existing variables are preserved, and the inner op's new variables are added alongside.

The merge is invoked with compat="no_conflicts" (xarray's safe default), so if the inner op produces a variable name that already exists in the input and the values differ, the merge raises rather than silently overwriting. Diagnostics that share an output name with one of their inputs (rare in practice) need to be renamed before being wrapped.

The combinator is the canonical idiom for chaining diagnostics inside a Sequential: each step augments the threaded Dataset with new columns, so later diagnostics can depend on variables added by earlier ones.

Parameters:

Name	Type	Description	Default
`inner`	`Operator`	The inner :class:`Operator`. Must accept a single `xr.Dataset` and return an `xr.Dataset`. Standard single-input `xrops.ocn` diagnostics (`RelativeVorticity`, `KineticEnergy`, `GeostrophicVelocities`, `Divergence`, ...) satisfy this contract. Two-input metrics (`RMSE`, `MAE`, ...) and operators that return a `DataArray` rather than a `Dataset` do not fit and will raise at call time.	required

Raises:

Type	Description
`TypeError`	If `inner` is not an :class:`Operator` (raised at construction), or if `inner(ds)` does not return a `Dataset` (raised at call time).
`MergeError`	If the inner op's output collides with an existing input variable whose values do not match.

Example

Build a single Dataset carrying every kinematic diagnostic for a velocity field::

from xrops import Augment, Sequential
from xrops.ocn.operators import (
    RelativeVorticity, KineticEnergy, OkuboWeiss,
)

diagnostics = Sequential([
    Augment(RelativeVorticity()),    # adds vort_r
    Augment(KineticEnergy()),         # adds ke
    Augment(OkuboWeiss()),            # adds ow
])
enriched = diagnostics(velocity_dataset)
# enriched has u, v, vort_r, ke, ow

Diagnostics that depend on previously-added columns work transparently because Sequential threads the merged Dataset forward::

Sequential([
    Augment(RelativeVorticity()),                # adds vort_r
    Augment(Enstrophy(variable="vort_r")),       # reads vort_r
])

ApplyToEach ¶

ApplyToEach(prototype: Operator, *, kwarg: str, values: Sequence[Any])

Bases: Operator

Re-instantiate a prototype operator per value and merge results.

Given a prototype :class:Operator, a kwarg name, and a sequence of values, :class:ApplyToEach builds and runs one new instance of the prototype's class per value, varying the chosen kwarg. Each resulting Dataset is merged into the input with the same safe compat="no_conflicts" semantics as :class:Augment.

The behaviour is equivalent to::

Sequential([
    Augment(prototype.__class__(**{**proto_config, kwarg: v}))
    for v in values
])

Each rebuilt operator runs against the threaded result, not the original input — i.e. the second pass sees the variables added by the first pass. This is what makes the equivalence with Sequential([Augment(...) for v in values]) exact, and lets later sweeps consume variables produced by earlier ones.

The prototype's get_config() is read once and used as the base kwargs dict for the rebuilt instances. The prototype's class must therefore satisfy the leaf contract cls(**proto.get_config()) — i.e. its get_config must return a dict directly usable as constructor kwargs. All single-operator Layer 1 classes in xrops satisfy this; the composite/combinator classes (Sequential, Graph, Augment, Tap, ApplyToEach itself) do not, because their configs carry serialized child records rather than live operator instances. ApplyToEach validates this contract at construction time via a sentinel rebuild and raises TypeError with a descriptive message if it fails.

Each rebuilt operator must produce uniquely-named output variables across the value sweep, otherwise the eventual merge raises. In practice most diagnostics name their output after the input variable they consume (e.g. Frontogenesis(scalar="ssh") → ssh_frontogenesis), so this is not usually a concern.

Parameters:

Name	Type	Description	Default
`prototype`	`Operator`	A configured single-operator :class:`Operator` instance. Composite classes such as `Sequential` or other combinators are not accepted (see contract above).	required
`kwarg`	`str`	Name of the constructor kwarg to vary across the sweep. Must be a key of `prototype.get_config()`.	required
`values`	`Sequence[Any]`	Sequence of values to substitute for `kwarg`. Each entry is used to build one rebuilt operator.	required

Raises:

Type	Description
`TypeError`	If `prototype` is not an :class:`Operator`, or if `prototype.get_config()` cannot be passed back to its class constructor (composite-config case), or if any rebuilt op does not return a Dataset.
`ValueError`	If `kwarg` is not present in `prototype.get_config()`.

Example

Compute frontogenesis for three scalars in one call::

from xrops import ApplyToEach
from xrops.ocn.operators import Frontogenesis

multi_fg = ApplyToEach(
    Frontogenesis(scalar="ssh"),
    kwarg="scalar",
    values=["ssh", "thetao", "so"],
)
out = multi_fg(ds)
# out has ssh_frontogenesis, thetao_frontogenesis,
# so_frontogenesis merged in alongside the original
# variables.

Inside a Sequential that augments the dataset before and after::

Sequential([
    ValidateCoords(),
    ApplyToEach(
        Frontogenesis(scalar="ssh"),
        kwarg="scalar",
        values=["thetao", "so"],
    ),
    Augment(KineticEnergy()),
])

Shape signatures¶

Signature is pipekit's shape descriptor (ordered (name, size) dims and a dtype), used by compute_output_signature for static shape inference across a pipeline.

Signature `dataclass` ¶

Signature(dims: tuple[tuple[str, int | None], ...], dtype: str | None = None)

Immutable shape + dtype record.

Attributes:

Name	Type	Description
`dims`	`tuple[tuple[str, int \| None], ...]`	Tuple of `(name, size)` pairs. `size` may be `None` to mark a dim as dynamic / unknown. Order matters — it's the carrier's axis order.
`dtype`	`str \| None`	A dtype label (e.g. `"float32"`, `"int64"`) or `None` if the operator doesn't track dtype.

Example::

sig = Signature(dims=(("time", 12), ("y", 256), ("x", 256)), dtype="float32")
sig.format()                       # "(time=12, y=256, x=256) float32"
sig.drop_dims(("time",))           # drops the time axis
sig.replace_dims({"x": 128})       # change the x size

format ¶

format() -> str

Return a compact "(name=size, …) dtype" string.

Source code in .venv/lib/python3.12/site-packages/pipekit/signature.py

def format(self) -> str:
    """Return a compact ``"(name=size, …) dtype"`` string."""
    parts = ", ".join(
        f"{name}={'?' if size is None else size}" for name, size in self.dims
    )
    dtype = self.dtype if self.dtype is not None else "?"
    return f"({parts}) {dtype}"

drop_dims ¶

drop_dims(names: tuple[str, ...]) -> Signature

Return a new Signature with the given dim names removed.

Raises:

Type	Description
`KeyError`	if a name in `names` isn't present.

Source code in .venv/lib/python3.12/site-packages/pipekit/signature.py

def drop_dims(self, names: tuple[str, ...]) -> Signature:
    """Return a new `Signature` with the given dim names removed.

    Raises:
        KeyError: if a name in ``names`` isn't present.
    """
    present = {n for n, _ in self.dims}
    missing = set(names) - present
    if missing:
        raise KeyError(f"drop_dims: unknown dim(s) {sorted(missing)}.")
    kept = tuple((n, s) for n, s in self.dims if n not in set(names))
    return Signature(dims=kept, dtype=self.dtype)

replace_dims ¶

replace_dims(sizes: dict[str, int | None]) -> Signature

Return a new Signature with the listed dim sizes overridden.

Raises:

Type	Description
`KeyError`	if a name in `sizes` isn't present.

Source code in .venv/lib/python3.12/site-packages/pipekit/signature.py

def replace_dims(self, sizes: dict[str, int | None]) -> Signature:
    """Return a new `Signature` with the listed dim sizes overridden.

    Raises:
        KeyError: if a name in ``sizes`` isn't present.
    """
    present = {n for n, _ in self.dims}
    missing = set(sizes) - present
    if missing:
        raise KeyError(f"replace_dims: unknown dim(s) {sorted(missing)}.")
    new = tuple((n, sizes.get(n, s)) for n, s in self.dims)
    return Signature(dims=new, dtype=self.dtype)

matches ¶

matches(pattern: str) -> bool

Whether these dims fit a dims pattern; sizes are ignored.

A pattern is whitespace-separated dim names with at most one ..., which matches zero or more dims at its position. Names compare exactly and in order.

Parameters:

Name	Type	Description	Default
`pattern`	`str`	E.g. `"band y x"` (exactly these dims) or `"... y x"` (any leading dims, then `y` and `x`).	required

Raises:

Type	Description
`ValueError`	if the pattern has more than one `...`.

Examples:

>>> sig = Signature(dims=(("time", 12), ("band", 2), ("y", 4), ("x", 4)))
>>> sig.matches("time band y x"), sig.matches("... y x")
(True, True)
>>> sig.matches("band y x"), sig.matches("time ... x")
(False, True)

Source code in .venv/lib/python3.12/site-packages/pipekit/signature.py

def matches(self, pattern: str) -> bool:
    """Whether these dims fit a dims pattern; sizes are ignored.

    A pattern is whitespace-separated dim names with at most one
    ``...``, which matches zero or more dims at its position. Names
    compare exactly and in order.

    Args:
        pattern: E.g. ``"band y x"`` (exactly these dims) or
            ``"... y x"`` (any leading dims, then ``y`` and ``x``).

    Raises:
        ValueError: if the pattern has more than one ``...``.

    Examples:
        >>> sig = Signature(dims=(("time", 12), ("band", 2), ("y", 4), ("x", 4)))
        >>> sig.matches("time band y x"), sig.matches("... y x")
        (True, True)
        >>> sig.matches("band y x"), sig.matches("time ... x")
        (False, True)
    """
    names, ellipsis = _parse_pattern(pattern)
    dims = tuple(name for name, _ in self.dims)
    if ellipsis is None:
        return dims == tuple(names)
    head = tuple(names[:ellipsis])
    tail = tuple(names[ellipsis:])
    if len(dims) < len(head) + len(tail):
        return False
    return dims[: len(head)] == head and dims[len(dims) - len(tail) :] == tail

from_pattern `classmethod` ¶

from_pattern(pattern: str, dtype: str | None = None) -> Signature

A sizeless Signature from a dims pattern, for annotations.

Parameters:

Name	Type	Description	Default
`pattern`	`str`	Whitespace-separated dim names, e.g. `"band y x"`.	required
`dtype`	`str \| None`	An optional dtype label.	`None`

Raises:

Type	Description
`ValueError`	if the pattern has a `...` (a signature needs concrete names) or repeats a name.

Example

Signature.from_pattern("band y x").dims (('band', None), ('y', None), ('x', None))

Source code in .venv/lib/python3.12/site-packages/pipekit/signature.py

@classmethod
def from_pattern(cls, pattern: str, dtype: str | None = None) -> Signature:
    """A sizeless `Signature` from a dims pattern, for annotations.

    Args:
        pattern: Whitespace-separated dim names, e.g. ``"band y x"``.
        dtype: An optional dtype label.

    Raises:
        ValueError: if the pattern has a ``...`` (a signature needs
            concrete names) or repeats a name.

    Example:
        >>> Signature.from_pattern("band y x").dims
        (('band', None), ('y', None), ('x', None))
    """
    names, ellipsis = _parse_pattern(pattern)
    if ellipsis is not None:
        raise ValueError(
            f"from_pattern needs concrete dim names, got {pattern!r} with '...'."
        )
    return cls(dims=tuple((name, None) for name in names), dtype=dtype)

Composition¶

Operator base class¶

Operator ¶

Sequential pipelines¶

Sequential ¶

compute_output_signature ¶

describe ¶

summary ¶

Functional Graph API¶

Input ¶

Node ¶

Graph ¶

get_config ¶

from_config classmethod ¶

compute_output_signature ¶

output_signatures ¶

summary ¶

describe ¶

Observation taps¶

Tap ¶

xarray-Dataset combinators¶

Augment ¶

ApplyToEach ¶

Shape signatures¶

Signature dataclass ¶

format ¶

drop_dims ¶

replace_dims ¶

matches ¶

from_pattern classmethod ¶

from_config `classmethod` ¶

Signature `dataclass` ¶

from_pattern `classmethod` ¶