Pipeline Recipes API

The wandas.pipeline module provides portable Recipe extraction, validation, serialization, replay, and extension declarations. For a guided executable lesson, use the Reusable Pipeline Recipes Learning Path. For task-oriented procedures, use the RecipePlan How-to.

段階的に学ぶ実行可能な教材は Reusable Pipeline Recipes Learning Pathへ、 具体的な作業手順はRecipePlan How-toへ進んでください。

wandas.pipeline

Portable semantic Recipe plans.

Attributes

__all__ = ['OperationCapture', 'RecipeExecutionError', 'RecipeExtractionError', 'RecipeOperation', 'RecipeOperationDescription', 'RecipeParameterDescription', 'RecipePlan', 'RecipeRegistry', 'RecipeSerializationError', 'RecipeValidationError', 'default_recipe_registry', 'recipe_definition', 'recipe_operation'] module-attribute

Classes

OperationCapture dataclass

Semantic inputs and parameters captured at a public method boundary.

Parameters:

Name	Type	Description	Default
bindings	tuple[InputBinding, ...]	Ordered input roles and runtime kinds for this invocation.	required
parents	tuple[LineageNode | None, ...]	Lineage parent for each binding; array bindings use None.	required
params	Mapping[str, Any]	Call intent to snapshot into semantic lineage. Nonportable top-level values become display-only markers and make Recipe extraction fail.	required
recipe_error	str | None	Optional explanation that makes later Recipe extraction fail atomically while retaining display history for the public call.	None

Source code in wandas/pipeline/decorators.py

@dataclass(frozen=True)
class OperationCapture:
    """Semantic inputs and parameters captured at a public method boundary.

    Args:
        bindings: Ordered input roles and runtime kinds for this invocation.
        parents: Lineage parent for each binding; array bindings use ``None``.
        params: Call intent to snapshot into semantic lineage. Nonportable top-level
            values become display-only markers and make Recipe extraction fail.
        recipe_error: Optional explanation that makes later Recipe extraction fail
            atomically while retaining display history for the public call.
    """

    bindings: tuple[InputBinding, ...]
    parents: tuple[LineageNode | None, ...]
    params: Mapping[str, Any]
    recipe_error: str | None = None

Attributes

bindings instance-attribute

parents instance-attribute

params instance-attribute

recipe_error = None class-attribute instance-attribute

Functions

__init__(bindings, parents, params, recipe_error=None)

RecipeExecutionError

Bases: RuntimeError

Raised when a validated Recipe cannot be applied to runtime inputs.

Attributes:

Name	Type	Description
node_id		Id of the Recipe node that failed, or None when the error is not tied to one node (for example, mismatched input names).
operation		Registered operation id of that node (RecipeNode.operation, for example "wandas.audio.lowpass_filter"), or None.
cause		The exception raised by the operation, the same object as __cause__, or None when the node failed a Recipe check itself. Like __cause__, it is not pickled, so it is None after the error crosses a process boundary; node_id, operation and the message are kept.

Source code in wandas/pipeline/errors.py

class RecipeExecutionError(RuntimeError):
    """Raised when a validated Recipe cannot be applied to runtime inputs.

    Attributes:
        node_id: Id of the Recipe node that failed, or ``None`` when the error is
            not tied to one node (for example, mismatched input names).
        operation: Registered operation id of that node (``RecipeNode.operation``,
            for example ``"wandas.audio.lowpass_filter"``), or ``None``.
        cause: The exception raised by the operation, the same object as
            ``__cause__``, or ``None`` when the node failed a Recipe check itself.
            Like ``__cause__``, it is not pickled, so it is ``None`` after the error
            crosses a process boundary; ``node_id``, ``operation`` and the message
            are kept.
    """

    def __init__(
        self,
        message: str,
        *,
        node_id: str | None = None,
        operation: str | None = None,
        cause: BaseException | None = None,
    ) -> None:
        super().__init__(message)
        self.node_id = node_id
        self.operation = operation
        self.cause = cause

    def __reduce__(self) -> tuple[Any, ...]:
        # The cause may not survive pickling (an exception whose __init__ does not take
        # its args), so leave it out like __cause__.
        return (
            _rebuild_recipe_execution_error,
            (type(self), self.args, self.node_id, self.operation),
        )

Attributes

node_id = node_id instance-attribute

operation = operation instance-attribute

cause = cause instance-attribute

Functions

__init__(message, *, node_id=None, operation=None, cause=None)

Source code in wandas/pipeline/errors.py

def __init__(
    self,
    message: str,
    *,
    node_id: str | None = None,
    operation: str | None = None,
    cause: BaseException | None = None,
) -> None:
    super().__init__(message)
    self.node_id = node_id
    self.operation = operation
    self.cause = cause

__reduce__()

Source code in wandas/pipeline/errors.py

def __reduce__(self) -> tuple[Any, ...]:
    # The cause may not survive pickling (an exception whose __init__ does not take
    # its args), so leave it out like __cause__.
    return (
        _rebuild_recipe_execution_error,
        (type(self), self.args, self.node_id, self.operation),
    )

RecipeExtractionError

Bases: ValueError

Raised when a frame lineage cannot be represented by a recipe.

Source code in wandas/pipeline/errors.py

class RecipeExtractionError(ValueError):
    """Raised when a frame lineage cannot be represented by a recipe."""

RecipeSerializationError

Bases: ValueError

Raised when a Recipe payload violates the canonical schema.

Source code in wandas/pipeline/errors.py

class RecipeSerializationError(ValueError):
    """Raised when a Recipe payload violates the canonical schema."""

RecipeValidationError

Bases: ValueError

Raised when a Recipe graph or registered operation contract is invalid.

Source code in wandas/pipeline/errors.py

class RecipeValidationError(ValueError):
    """Raised when a Recipe graph or registered operation contract is invalid."""

RecipePlan dataclass

Portable, validated graph of public Wandas Frame operations.

A plan stores operation intent and named input slots, never Frame samples or a Dask task graph. Constructing, serializing, loading, and applying a plan therefore remain lazy; computation occurs only when the returned Frame is computed.

User workflows create plans with :meth:from_frame or :meth:from_dict, then use :meth:apply and :meth:to_dict. The direct inputs/nodes/output constructor is an internal graph-assembly interface, not a public compatibility surface.

A plan returns one Frame per :meth:apply. Besides its output, it can name other steps, such as a shared preprocessing result or a second analysis of it; apply(..., step=name) returns that step instead and builds only the operations it needs.

Parameters:

Name	Type	Description	Default
inputs	Iterable[RecipeInput]	Named Frame or NumPy/Dask array inputs used by the graph.	required
nodes	Iterable[RecipeNode]	Topologically ordered Recipe operations.	required
output	str	Identifier of the Frame input or node returned by the plan.	required
steps	Iterable[tuple[str, str]]	(name, identifier) pairs naming other Frames apply can return, in order.	()
registry	RecipeRegistry | None	Immutable operation registry used to validate the graph. Uses the built-in registry when omitted.	None

Raises:

Type	Description
RecipeValidationError	If the graph or an operation contract is invalid.

Source code in wandas/pipeline/model.py

@dataclass(frozen=True)
class RecipePlan:
    """Portable, validated graph of public Wandas Frame operations.

    A plan stores operation intent and named input slots, never Frame samples or a
    Dask task graph. Constructing, serializing, loading, and applying a plan therefore
    remain lazy; computation occurs only when the returned Frame is computed.

    User workflows create plans with :meth:`from_frame` or :meth:`from_dict`, then use
    :meth:`apply` and :meth:`to_dict`. The direct ``inputs``/``nodes``/``output``
    constructor is an internal graph-assembly interface, not a public compatibility
    surface.

    A plan returns one Frame per :meth:`apply`. Besides its output, it can name
    other steps, such as a shared preprocessing result or a second analysis of it;
    ``apply(..., step=name)`` returns that step instead and builds only the
    operations it needs.

    Args:
        inputs: Named Frame or NumPy/Dask array inputs used by the graph.
        nodes: Topologically ordered Recipe operations.
        output: Identifier of the Frame input or node returned by the plan.
        steps: ``(name, identifier)`` pairs naming other Frames ``apply`` can
            return, in order.
        registry: Immutable operation registry used to validate the graph. Uses the
            built-in registry when omitted.

    Raises:
        RecipeValidationError: If the graph or an operation contract is invalid.
    """

    inputs: tuple[RecipeInput, ...]
    nodes: tuple[RecipeNode, ...]
    output: str
    steps: tuple[tuple[str, str], ...]

    def __init__(
        self,
        inputs: Iterable[RecipeInput],
        nodes: Iterable[RecipeNode],
        output: str,
        *,
        steps: Iterable[tuple[str, str]] = (),
        registry: RecipeRegistry | None = None,
    ) -> None:
        """Snapshot and validate a complete Recipe graph."""
        object.__setattr__(self, "inputs", tuple(inputs))
        object.__setattr__(self, "nodes", tuple(nodes))
        object.__setattr__(self, "output", output)
        object.__setattr__(self, "steps", tuple(tuple(step) for step in steps))
        validate_recipe_plan(self, registry=registry)

    @property
    def step_names(self) -> tuple[str, ...]:
        """Names accepted by ``apply(..., step=...)``, in order."""
        return tuple(name for name, _reference in self.steps)

    def for_step(self, step: str, *, registry: RecipeRegistry | None = None) -> RecipePlan:
        """Return a plan whose output is the named step, without other steps.

        The new plan keeps only the operations and inputs that step needs, so it can
        be saved or shared on its own. Applied to those inputs, it returns the same
        Frame as ``plan.apply(..., step=name)``, which still takes every input of
        this plan.

        Args:
            step: Name from :attr:`step_names`.
            registry: Registry used to validate the new plan, as in :meth:`apply`.

        Returns:
            A validated plan with the inputs that step uses and no steps.

        Raises:
            RecipeExecutionError: If ``step`` is not one of :attr:`step_names`.
        """
        target = _step_reference(self, step)
        required = _required_references(self, {target})
        return RecipePlan(
            (item for item in self.inputs if item.id in required),
            (node for node in self.nodes if node.id in required),
            target,
            registry=registry,
        )

    @classmethod
    def from_frame(
        cls,
        frame: Any,
        *,
        input_names: tuple[str, ...] | None = None,
        steps: Mapping[str, Any] | None = None,
        registry: RecipeRegistry | None = None,
    ) -> RecipePlan:
        """Extract a Recipe from a Frame's semantic lineage.

        Args:
            frame: Result Frame produced through public Recipe-capable operations.
            input_names: Optional names assigned to discovered runtime inputs in
                deterministic traversal order. When omitted, names are generated as
                ``input_0``, ``input_1``, and so on.
            steps: Other Frames to name in the plan, by name, so that
                ``apply(..., step=name)`` returns them. They must come from the same
                source Frames as ``frame``; operations they share with ``frame``
                are stored once.
            registry: Registry used to resolve every captured operation. Uses the
                built-in registry when omitted.

        Returns:
            A validated plan whose output reproduces ``frame``'s public workflow.

        Raises:
            RecipeExtractionError: If the value is not a Frame, an operation is not
                portable or registered, ``input_names`` has the wrong length, or a
                step is not a Frame, has an invalid name, or uses another source.

        Examples:
            >>> import wandas as wd
            >>> from wandas.pipeline import RecipePlan
            >>> signal = wd.generate_sin(freqs=[1000.0], duration=0.1)
            >>> weighted = signal.high_pass_filter(cutoff=50).a_weighting()
            >>> plan = RecipePlan.from_frame(
            ...     weighted.welch(), steps={"weighted": weighted, "noct": weighted.noct_spectrum()}
            ... )
            >>> plan.step_names
            ('weighted', 'noct')
            >>> type(plan.apply(signal, step="noct")).__name__
            'NOctFrame'
        """
        from wandas.pipeline.compiler import LineageRecipeCompiler

        return LineageRecipeCompiler(input_names=input_names, registry=registry).compile_frame(frame, steps=steps)

    @classmethod
    def from_dict(
        cls,
        payload: Mapping[str, Any],
        *,
        registry: RecipeRegistry | None = None,
    ) -> RecipePlan:
        """Load and validate a Recipe schema-2 or schema-3 mapping.

        Args:
            payload: Mapping produced by :meth:`to_dict` or an equivalent decoded
                JSON object.
            registry: Registry used to validate operation identifiers and versions.

        Returns:
            A new immutable Recipe plan.

        Raises:
            RecipeSerializationError: If the payload violates the schema or graph
                contract.
        """
        from wandas.pipeline.serialization import RecipeLoader

        return RecipeLoader(registry=registry).load(payload)

    def to_dict(self) -> dict[str, Any]:
        """Return the deterministic JSON-compatible Recipe schema.

        Returns:
            A fresh mapping using schema ``wandas.recipe`` version 3.
        """
        from wandas.pipeline.serialization import RecipeSerializer

        return RecipeSerializer().serialize(self)

    def save(self, path: str | Path, *, overwrite: bool = False) -> Path:
        """Save this plan as a standalone ``.recipe.json`` artifact.

        The artifact contains operation intent and named inputs only. Frame samples,
        WDF data, and Dask graphs remain outside the Recipe persistence boundary.

        Args:
            path: Target path. ``.recipe.json`` is appended when absent.
            overwrite: Replace an existing artifact when true.

        Returns:
            The normalized artifact path written to disk.
        """
        from wandas.pipeline.artifacts import save_recipe_artifact

        return save_recipe_artifact(self, path, overwrite=overwrite)

    @classmethod
    def load(
        cls,
        path: str | Path,
        *,
        registry: RecipeRegistry | None = None,
    ) -> RecipePlan:
        """Load and validate a standalone ``.recipe.json`` artifact.

        Args:
            path: Artifact path. ``.recipe.json`` is appended when absent.
            registry: Registry used to validate operation identifiers and versions.

        Returns:
            A validated immutable Recipe plan.

        Raises:
            FileNotFoundError: If the normalized artifact path does not exist.
            RecipeSerializationError: If decoding, schema validation, or graph
                validation fails.
        """
        from wandas.pipeline.artifacts import load_recipe_artifact

        return load_recipe_artifact(path, registry=registry)

    def apply(
        self,
        inputs: Mapping[str, Any] | Any,
        *,
        step: str | None = None,
        registry: RecipeRegistry | None = None,
    ) -> Any:
        """Build the Recipe output, or one named step, lazily from named runtime inputs.

        Args:
            inputs: Exact mapping from each :class:`RecipeInput` name to a Wandas
                Frame or NumPy/Dask array of the declared kind. A plan with one
                input also accepts that Frame or array directly, so
                ``plan.apply(frame)`` equals ``plan.apply({plan.inputs[0].name: frame})``.
            step: Name from :attr:`step_names` to return instead of the output. Only
                the operations that step depends on are built. ``None`` returns the
                output.
            registry: Registry used for validation and operation execution. Supply
                the same extension registry used during extraction and loading.

        Returns:
            The output Frame (or the named step) with semantic lineage, metadata, and
                Dask laziness preserved by its public operations. An identity plan with
                no nodes returns its input Frame unchanged.

        Raises:
            RecipeExecutionError: If names or runtime value kinds do not match, a
                plan with several inputs receives one value instead of a mapping,
                ``step`` is not one of :attr:`step_names`, an operation fails, or an
                operation violates the Frame lineage contract.
            RecipeValidationError: If the plan is invalid for ``registry``.
        """
        if not isinstance(inputs, Mapping):
            if len(self.inputs) != 1:
                from wandas.pipeline.errors import RecipeExecutionError

                raise RecipeExecutionError(
                    "Recipe with several inputs requires a mapping by input name\n"
                    f"  Inputs: {[recipe_input.name for recipe_input in self.inputs]!r}\n"
                    f"  Got: {type(inputs).__name__}"
                )
            inputs = {self.inputs[0].name: inputs}
        return RecipeExecutor(registry=registry).execute(self, inputs, step=step)

    def apply_files(
        self,
        paths: Iterable[str | Path],
        out_dir: str | Path,
        *,
        root: str | Path | None = None,
        jobs: int = 1,
        on_error: Literal["continue", "raise"] = "continue",
        overwrite: bool = False,
        step: str | None = None,
        registry: RecipeRegistry | None = None,
    ) -> Any:
        """Apply this one-input plan to many recordings and save each result as WDF.

        Each recording is read with :func:`wandas.read`, passed to :meth:`apply`, and
        saved with ``save`` to ``out_dir/<name>.wdf``. Reading uses the dtype of
        :func:`wandas.get_default_dtype` at this call, in worker processes too. The name keeps the recording's
        extension (``a.wav`` becomes ``a.wav.wdf``), so ``a.wav`` and ``a.flac`` do not
        share an output. With ``root``, the path below ``root`` is kept, so
        ``root/L1/a.wav`` becomes ``out_dir/L1/a.wav.wdf``. Saving needs the ``io``
        extra (``pip install "wandas[io]"``).

        Args:
            paths: Local recording paths, in the order of the returned rows.
            out_dir: Folder for the outputs; created as needed.
            root: Folder the paths are below; keeps their relative folders in
                ``out_dir``. Without it, outputs are named by file name only.
            jobs: Recordings processed at the same time, in separate processes.
                Workers are started with ``spawn``, so a script that uses
                ``jobs > 1`` needs the usual ``if __name__ == "__main__":`` guard.
            on_error: ``"continue"`` records a failed recording and goes on;
                ``"raise"`` stops at the first failure and raises its exception. With
                ``jobs > 1``, recordings after it may already be running or saved.
                A worker process that dies fails its recording and every recording
                not finished yet.
            overwrite: Replace existing outputs. Without it, an existing output is
                that recording's failure.
            step: Name from :attr:`step_names` to save instead of the output, as
                in :meth:`apply`.
            registry: Registry for validation and execution, as in :meth:`apply`.
                Only with ``jobs=1``, since a registry cannot be sent to processes.

        Returns:
            A pandas ``DataFrame`` with one row per recording and the columns
                ``path``, ``status`` (``"ok"`` or ``"error"``), ``output`` (empty on
                error), ``error`` (exception type and first message line, prefixed
                with the operation id when an operation failed; empty on success),
                and ``seconds``.

        Raises:
            RecipeExecutionError: If the plan does not have exactly one input.
            RecipeExecutionError: If ``step`` is not one of :attr:`step_names`.
            ValueError: If ``jobs`` or ``on_error`` is invalid, a custom
                ``registry`` is combined with ``jobs > 1``, a path is not below
                ``root``, or two recordings would share an output.
            Exception: With ``on_error="raise"``, the first recording's failure.
        """
        from wandas.pipeline.batch import apply_files

        return apply_files(
            self,
            paths,
            out_dir,
            root=root,
            jobs=jobs,
            on_error=on_error,
            overwrite=overwrite,
            step=step,
            registry=registry,
        )

Attributes

inputs instance-attribute

nodes instance-attribute

output instance-attribute

steps instance-attribute

step_names property

Names accepted by apply(..., step=...), in order.

Functions

__init__(inputs, nodes, output, *, steps=(), registry=None)

Snapshot and validate a complete Recipe graph.

Source code in wandas/pipeline/model.py

def __init__(
    self,
    inputs: Iterable[RecipeInput],
    nodes: Iterable[RecipeNode],
    output: str,
    *,
    steps: Iterable[tuple[str, str]] = (),
    registry: RecipeRegistry | None = None,
) -> None:
    """Snapshot and validate a complete Recipe graph."""
    object.__setattr__(self, "inputs", tuple(inputs))
    object.__setattr__(self, "nodes", tuple(nodes))
    object.__setattr__(self, "output", output)
    object.__setattr__(self, "steps", tuple(tuple(step) for step in steps))
    validate_recipe_plan(self, registry=registry)

for_step(step, *, registry=None)

Return a plan whose output is the named step, without other steps.

The new plan keeps only the operations and inputs that step needs, so it can be saved or shared on its own. Applied to those inputs, it returns the same Frame as plan.apply(..., step=name), which still takes every input of this plan.

Parameters:

Name	Type	Description	Default
step	str	Name from :attr:step_names.	required
registry	RecipeRegistry | None	Registry used to validate the new plan, as in :meth:apply.	None

Returns:

Type	Description
RecipePlan	A validated plan with the inputs that step uses and no steps.

Raises:

Type	Description
RecipeExecutionError	If step is not one of :attr:step_names.

Source code in wandas/pipeline/model.py

def for_step(self, step: str, *, registry: RecipeRegistry | None = None) -> RecipePlan:
    """Return a plan whose output is the named step, without other steps.

    The new plan keeps only the operations and inputs that step needs, so it can
    be saved or shared on its own. Applied to those inputs, it returns the same
    Frame as ``plan.apply(..., step=name)``, which still takes every input of
    this plan.

    Args:
        step: Name from :attr:`step_names`.
        registry: Registry used to validate the new plan, as in :meth:`apply`.

    Returns:
        A validated plan with the inputs that step uses and no steps.

    Raises:
        RecipeExecutionError: If ``step`` is not one of :attr:`step_names`.
    """
    target = _step_reference(self, step)
    required = _required_references(self, {target})
    return RecipePlan(
        (item for item in self.inputs if item.id in required),
        (node for node in self.nodes if node.id in required),
        target,
        registry=registry,
    )

from_frame(frame, *, input_names=None, steps=None, registry=None) classmethod

Extract a Recipe from a Frame's semantic lineage.

Parameters:

Name	Type	Description	Default
frame	Any	Result Frame produced through public Recipe-capable operations.	required
input_names	tuple[str, ...] | None	Optional names assigned to discovered runtime inputs in deterministic traversal order. When omitted, names are generated as input_0, input_1, and so on.	None
steps	Mapping[str, Any] | None	Other Frames to name in the plan, by name, so that apply(..., step=name) returns them. They must come from the same source Frames as frame; operations they share with frame are stored once.	None
registry	RecipeRegistry | None	Registry used to resolve every captured operation. Uses the built-in registry when omitted.	None

Returns:

Type	Description
RecipePlan	A validated plan whose output reproduces frame's public workflow.

Raises:

Type	Description
RecipeExtractionError	If the value is not a Frame, an operation is not portable or registered, input_names has the wrong length, or a step is not a Frame, has an invalid name, or uses another source.

Examples:

>>> import wandas as wd
>>> from wandas.pipeline import RecipePlan
>>> signal = wd.generate_sin(freqs=[1000.0], duration=0.1)
>>> weighted = signal.high_pass_filter(cutoff=50).a_weighting()
>>> plan = RecipePlan.from_frame(
...     weighted.welch(), steps={"weighted": weighted, "noct": weighted.noct_spectrum()}
... )
>>> plan.step_names
('weighted', 'noct')
>>> type(plan.apply(signal, step="noct")).__name__
'NOctFrame'

Source code in wandas/pipeline/model.py

@classmethod
def from_frame(
    cls,
    frame: Any,
    *,
    input_names: tuple[str, ...] | None = None,
    steps: Mapping[str, Any] | None = None,
    registry: RecipeRegistry | None = None,
) -> RecipePlan:
    """Extract a Recipe from a Frame's semantic lineage.

    Args:
        frame: Result Frame produced through public Recipe-capable operations.
        input_names: Optional names assigned to discovered runtime inputs in
            deterministic traversal order. When omitted, names are generated as
            ``input_0``, ``input_1``, and so on.
        steps: Other Frames to name in the plan, by name, so that
            ``apply(..., step=name)`` returns them. They must come from the same
            source Frames as ``frame``; operations they share with ``frame``
            are stored once.
        registry: Registry used to resolve every captured operation. Uses the
            built-in registry when omitted.

    Returns:
        A validated plan whose output reproduces ``frame``'s public workflow.

    Raises:
        RecipeExtractionError: If the value is not a Frame, an operation is not
            portable or registered, ``input_names`` has the wrong length, or a
            step is not a Frame, has an invalid name, or uses another source.

    Examples:
        >>> import wandas as wd
        >>> from wandas.pipeline import RecipePlan
        >>> signal = wd.generate_sin(freqs=[1000.0], duration=0.1)
        >>> weighted = signal.high_pass_filter(cutoff=50).a_weighting()
        >>> plan = RecipePlan.from_frame(
        ...     weighted.welch(), steps={"weighted": weighted, "noct": weighted.noct_spectrum()}
        ... )
        >>> plan.step_names
        ('weighted', 'noct')
        >>> type(plan.apply(signal, step="noct")).__name__
        'NOctFrame'
    """
    from wandas.pipeline.compiler import LineageRecipeCompiler

    return LineageRecipeCompiler(input_names=input_names, registry=registry).compile_frame(frame, steps=steps)

from_dict(payload, *, registry=None) classmethod

Load and validate a Recipe schema-2 or schema-3 mapping.

Parameters:

Name	Type	Description	Default
payload	Mapping[str, Any]	Mapping produced by :meth:to_dict or an equivalent decoded JSON object.	required
registry	RecipeRegistry | None	Registry used to validate operation identifiers and versions.	None

Returns:

Type	Description
RecipePlan	A new immutable Recipe plan.

Raises:

Type	Description
RecipeSerializationError	If the payload violates the schema or graph contract.

Source code in wandas/pipeline/model.py

@classmethod
def from_dict(
    cls,
    payload: Mapping[str, Any],
    *,
    registry: RecipeRegistry | None = None,
) -> RecipePlan:
    """Load and validate a Recipe schema-2 or schema-3 mapping.

    Args:
        payload: Mapping produced by :meth:`to_dict` or an equivalent decoded
            JSON object.
        registry: Registry used to validate operation identifiers and versions.

    Returns:
        A new immutable Recipe plan.

    Raises:
        RecipeSerializationError: If the payload violates the schema or graph
            contract.
    """
    from wandas.pipeline.serialization import RecipeLoader

    return RecipeLoader(registry=registry).load(payload)

to_dict()

Return the deterministic JSON-compatible Recipe schema.

Returns:

Type	Description
dict[str, Any]	A fresh mapping using schema wandas.recipe version 3.

Source code in wandas/pipeline/model.py

def to_dict(self) -> dict[str, Any]:
    """Return the deterministic JSON-compatible Recipe schema.

    Returns:
        A fresh mapping using schema ``wandas.recipe`` version 3.
    """
    from wandas.pipeline.serialization import RecipeSerializer

    return RecipeSerializer().serialize(self)

save(path, *, overwrite=False)

Save this plan as a standalone .recipe.json artifact.

The artifact contains operation intent and named inputs only. Frame samples, WDF data, and Dask graphs remain outside the Recipe persistence boundary.

Parameters:

Name	Type	Description	Default
path	str | Path	Target path. .recipe.json is appended when absent.	required
overwrite	bool	Replace an existing artifact when true.	False

Returns:

Type	Description
Path	The normalized artifact path written to disk.

Source code in wandas/pipeline/model.py

def save(self, path: str | Path, *, overwrite: bool = False) -> Path:
    """Save this plan as a standalone ``.recipe.json`` artifact.

    The artifact contains operation intent and named inputs only. Frame samples,
    WDF data, and Dask graphs remain outside the Recipe persistence boundary.

    Args:
        path: Target path. ``.recipe.json`` is appended when absent.
        overwrite: Replace an existing artifact when true.

    Returns:
        The normalized artifact path written to disk.
    """
    from wandas.pipeline.artifacts import save_recipe_artifact

    return save_recipe_artifact(self, path, overwrite=overwrite)

load(path, *, registry=None) classmethod

Load and validate a standalone .recipe.json artifact.

Parameters:

Name	Type	Description	Default
path	str | Path	Artifact path. .recipe.json is appended when absent.	required
registry	RecipeRegistry | None	Registry used to validate operation identifiers and versions.	None

Returns:

Type	Description
RecipePlan	A validated immutable Recipe plan.

Raises:

Type	Description
FileNotFoundError	If the normalized artifact path does not exist.
RecipeSerializationError	If decoding, schema validation, or graph validation fails.

Source code in wandas/pipeline/model.py

@classmethod
def load(
    cls,
    path: str | Path,
    *,
    registry: RecipeRegistry | None = None,
) -> RecipePlan:
    """Load and validate a standalone ``.recipe.json`` artifact.

    Args:
        path: Artifact path. ``.recipe.json`` is appended when absent.
        registry: Registry used to validate operation identifiers and versions.

    Returns:
        A validated immutable Recipe plan.

    Raises:
        FileNotFoundError: If the normalized artifact path does not exist.
        RecipeSerializationError: If decoding, schema validation, or graph
            validation fails.
    """
    from wandas.pipeline.artifacts import load_recipe_artifact

    return load_recipe_artifact(path, registry=registry)

apply(inputs, *, step=None, registry=None)

Build the Recipe output, or one named step, lazily from named runtime inputs.

Parameters:

Name	Type	Description	Default
inputs	Mapping[str, Any] | Any	Exact mapping from each :class:RecipeInput name to a Wandas Frame or NumPy/Dask array of the declared kind. A plan with one input also accepts that Frame or array directly, so plan.apply(frame) equals plan.apply({plan.inputs[0].name: frame}).	required
step	str | None	Name from :attr:step_names to return instead of the output. Only the operations that step depends on are built. None returns the output.	None
registry	RecipeRegistry | None	Registry used for validation and operation execution. Supply the same extension registry used during extraction and loading.	None

Returns:

Type	Description
Any	The output Frame (or the named step) with semantic lineage, metadata, and Dask laziness preserved by its public operations. An identity plan with no nodes returns its input Frame unchanged.

Raises:

Type	Description
RecipeExecutionError	If names or runtime value kinds do not match, a plan with several inputs receives one value instead of a mapping, step is not one of :attr:step_names, an operation fails, or an operation violates the Frame lineage contract.
RecipeValidationError	If the plan is invalid for registry.

Source code in wandas/pipeline/model.py

def apply(
    self,
    inputs: Mapping[str, Any] | Any,
    *,
    step: str | None = None,
    registry: RecipeRegistry | None = None,
) -> Any:
    """Build the Recipe output, or one named step, lazily from named runtime inputs.

    Args:
        inputs: Exact mapping from each :class:`RecipeInput` name to a Wandas
            Frame or NumPy/Dask array of the declared kind. A plan with one
            input also accepts that Frame or array directly, so
            ``plan.apply(frame)`` equals ``plan.apply({plan.inputs[0].name: frame})``.
        step: Name from :attr:`step_names` to return instead of the output. Only
            the operations that step depends on are built. ``None`` returns the
            output.
        registry: Registry used for validation and operation execution. Supply
            the same extension registry used during extraction and loading.

    Returns:
        The output Frame (or the named step) with semantic lineage, metadata, and
            Dask laziness preserved by its public operations. An identity plan with
            no nodes returns its input Frame unchanged.

    Raises:
        RecipeExecutionError: If names or runtime value kinds do not match, a
            plan with several inputs receives one value instead of a mapping,
            ``step`` is not one of :attr:`step_names`, an operation fails, or an
            operation violates the Frame lineage contract.
        RecipeValidationError: If the plan is invalid for ``registry``.
    """
    if not isinstance(inputs, Mapping):
        if len(self.inputs) != 1:
            from wandas.pipeline.errors import RecipeExecutionError

            raise RecipeExecutionError(
                "Recipe with several inputs requires a mapping by input name\n"
                f"  Inputs: {[recipe_input.name for recipe_input in self.inputs]!r}\n"
                f"  Got: {type(inputs).__name__}"
            )
        inputs = {self.inputs[0].name: inputs}
    return RecipeExecutor(registry=registry).execute(self, inputs, step=step)

apply_files(paths, out_dir, *, root=None, jobs=1, on_error='continue', overwrite=False, step=None, registry=None)

Apply this one-input plan to many recordings and save each result as WDF.

Each recording is read with :func:wandas.read, passed to :meth:apply, and saved with save to out_dir/<name>.wdf. Reading uses the dtype of :func:wandas.get_default_dtype at this call, in worker processes too. The name keeps the recording's extension (a.wav becomes a.wav.wdf), so a.wav and a.flac do not share an output. With root, the path below root is kept, so root/L1/a.wav becomes out_dir/L1/a.wav.wdf. Saving needs the io extra (pip install "wandas[io]").

Parameters:

Name	Type	Description	Default
paths	Iterable[str | Path]	Local recording paths, in the order of the returned rows.	required
out_dir	str | Path	Folder for the outputs; created as needed.	required
root	str | Path | None	Folder the paths are below; keeps their relative folders in out_dir. Without it, outputs are named by file name only.	None
jobs	int	Recordings processed at the same time, in separate processes. Workers are started with spawn, so a script that uses jobs > 1 needs the usual if __name__ == "__main__": guard.	1
on_error	Literal['continue', 'raise']	"continue" records a failed recording and goes on; "raise" stops at the first failure and raises its exception. With jobs > 1, recordings after it may already be running or saved. A worker process that dies fails its recording and every recording not finished yet.	'continue'
overwrite	bool	Replace existing outputs. Without it, an existing output is that recording's failure.	False
step	str | None	Name from :attr:step_names to save instead of the output, as in :meth:apply.	None
registry	RecipeRegistry | None	Registry for validation and execution, as in :meth:apply. Only with jobs=1, since a registry cannot be sent to processes.	None

Returns:

Type	Description
Any	A pandas DataFrame with one row per recording and the columns path, status ("ok" or "error"), output (empty on error), error (exception type and first message line, prefixed with the operation id when an operation failed; empty on success), and seconds.

Raises:

Type	Description
RecipeExecutionError	If the plan does not have exactly one input.
RecipeExecutionError	If step is not one of :attr:step_names.
ValueError	If jobs or on_error is invalid, a custom registry is combined with jobs > 1, a path is not below root, or two recordings would share an output.
Exception	With on_error="raise", the first recording's failure.

Source code in wandas/pipeline/model.py

def apply_files(
    self,
    paths: Iterable[str | Path],
    out_dir: str | Path,
    *,
    root: str | Path | None = None,
    jobs: int = 1,
    on_error: Literal["continue", "raise"] = "continue",
    overwrite: bool = False,
    step: str | None = None,
    registry: RecipeRegistry | None = None,
) -> Any:
    """Apply this one-input plan to many recordings and save each result as WDF.

    Each recording is read with :func:`wandas.read`, passed to :meth:`apply`, and
    saved with ``save`` to ``out_dir/<name>.wdf``. Reading uses the dtype of
    :func:`wandas.get_default_dtype` at this call, in worker processes too. The name keeps the recording's
    extension (``a.wav`` becomes ``a.wav.wdf``), so ``a.wav`` and ``a.flac`` do not
    share an output. With ``root``, the path below ``root`` is kept, so
    ``root/L1/a.wav`` becomes ``out_dir/L1/a.wav.wdf``. Saving needs the ``io``
    extra (``pip install "wandas[io]"``).

    Args:
        paths: Local recording paths, in the order of the returned rows.
        out_dir: Folder for the outputs; created as needed.
        root: Folder the paths are below; keeps their relative folders in
            ``out_dir``. Without it, outputs are named by file name only.
        jobs: Recordings processed at the same time, in separate processes.
            Workers are started with ``spawn``, so a script that uses
            ``jobs > 1`` needs the usual ``if __name__ == "__main__":`` guard.
        on_error: ``"continue"`` records a failed recording and goes on;
            ``"raise"`` stops at the first failure and raises its exception. With
            ``jobs > 1``, recordings after it may already be running or saved.
            A worker process that dies fails its recording and every recording
            not finished yet.
        overwrite: Replace existing outputs. Without it, an existing output is
            that recording's failure.
        step: Name from :attr:`step_names` to save instead of the output, as
            in :meth:`apply`.
        registry: Registry for validation and execution, as in :meth:`apply`.
            Only with ``jobs=1``, since a registry cannot be sent to processes.

    Returns:
        A pandas ``DataFrame`` with one row per recording and the columns
            ``path``, ``status`` (``"ok"`` or ``"error"``), ``output`` (empty on
            error), ``error`` (exception type and first message line, prefixed
            with the operation id when an operation failed; empty on success),
            and ``seconds``.

    Raises:
        RecipeExecutionError: If the plan does not have exactly one input.
        RecipeExecutionError: If ``step`` is not one of :attr:`step_names`.
        ValueError: If ``jobs`` or ``on_error`` is invalid, a custom
            ``registry`` is combined with ``jobs > 1``, a path is not below
            ``root``, or two recordings would share an output.
        Exception: With ``on_error="raise"``, the first recording's failure.
    """
    from wandas.pipeline.batch import apply_files

    return apply_files(
        self,
        paths,
        out_dir,
        root=root,
        jobs=jobs,
        on_error=on_error,
        overwrite=overwrite,
        step=step,
        registry=registry,
    )

RecipeOperation dataclass

Complete versioned contract for one portable public operation.

Operation values use identity-based equality so independently declared handlers are never treated as interchangeable merely because their metadata matches.

Parameters:

Name	Type	Description	Default
operation_id	str	Stable identifier persisted in Recipe nodes.	required
version	int	Positive version of this operation contract.	required
binding_patterns	tuple[tuple[InputBinding, ...], ...]	Accepted ordered input-role and input-kind patterns. Runtime kind signatures must be unique.	required
handler	RecipeHandler	Replay callback receiving ordered runtime inputs and immutable params.	required
validate_params	ParamValidator	Callback that rejects invalid decoded parameters.	_no_param_validation
method	Callable[..., Any] | None	Decorated method whose calls record this operation. It is used only by :meth:describe; capture and replay never call it through this field.	None
extras	tuple[str, ...]	Optional wandas extras the operation needs, such as ("effects",). Reported by :meth:describe; not checked at replay.	()

Source code in wandas/pipeline/registry.py

@dataclass(frozen=True, eq=False)
class RecipeOperation:
    """Complete versioned contract for one portable public operation.

    Operation values use identity-based equality so independently declared handlers are
    never treated as interchangeable merely because their metadata matches.

    Args:
        operation_id: Stable identifier persisted in Recipe nodes.
        version: Positive version of this operation contract.
        binding_patterns: Accepted ordered input-role and input-kind patterns. Runtime
            kind signatures must be unique.
        handler: Replay callback receiving ordered runtime inputs and immutable params.
        validate_params: Callback that rejects invalid decoded parameters.
        method: Decorated method whose calls record this operation. It is used only by
            :meth:`describe`; capture and replay never call it through this field.
        extras: Optional ``wandas`` extras the operation needs, such as
            ``("effects",)``. Reported by :meth:`describe`; not checked at replay.
    """

    operation_id: str
    version: int
    binding_patterns: tuple[tuple[InputBinding, ...], ...]
    handler: RecipeHandler = field(repr=False)
    validate_params: ParamValidator = field(default=_no_param_validation, repr=False)
    method: Callable[..., Any] | None = field(default=None, repr=False)
    extras: tuple[str, ...] = ()

    def __post_init__(self) -> None:
        """Snapshot and validate the complete operation contract."""
        object.__setattr__(self, "binding_patterns", tuple(self.binding_patterns))
        if isinstance(self.extras, str):
            raise TypeError("Recipe operation extras must be a tuple of strings, not a str")
        object.__setattr__(self, "extras", tuple(self.extras))
        if any(not isinstance(extra, str) or not extra.strip() for extra in self.extras):
            raise ValueError("Recipe operation extras must be non-blank strings")
        if not isinstance(self.operation_id, str) or not self.operation_id.strip():
            raise ValueError("Recipe operation id must be a non-blank string")
        if type(self.version) is not int or self.version < 1:
            raise ValueError("Recipe operation version must be a positive integer")
        if not self.binding_patterns:
            raise ValueError("Recipe operation requires at least one binding pattern")
        if any(not pattern for pattern in self.binding_patterns):
            raise ValueError("Recipe binding patterns must each contain at least one input")
        if not all(
            isinstance(pattern, tuple) and all(isinstance(binding, InputBinding) for binding in pattern)
            for pattern in self.binding_patterns
        ):
            raise TypeError("Recipe binding patterns must contain InputBinding tuples")
        if len(set(self.binding_patterns)) != len(self.binding_patterns):
            raise ValueError("Recipe binding patterns must be unique")
        if any(len({binding.role for binding in pattern}) != len(pattern) for pattern in self.binding_patterns):
            raise ValueError("Recipe binding roles must be unique within each pattern")
        kind_patterns = tuple(tuple(binding.kind for binding in pattern) for pattern in self.binding_patterns)
        if len(set(kind_patterns)) != len(kind_patterns):
            raise ValueError("Recipe binding patterns must have unique input kind signatures")
        if not callable(self.handler) or not callable(self.validate_params):
            raise TypeError("Recipe handler and parameter validator must be callable")
        if self.method is not None and not callable(self.method):
            raise TypeError("Recipe operation method must be callable")

    def accepts(self, bindings: tuple[InputBinding, ...]) -> bool:
        """Return whether ``bindings`` exactly match a declared pattern."""
        return bindings in self.binding_patterns

    def invoke(self, inputs: tuple[Any, ...], params: FrozenMap) -> Any:
        """Invoke the replay handler after complete-plan validation.

        Args:
            inputs: Ordered runtime values matching one declared binding pattern.
            params: Canonical parameters already accepted by ``validate_params``.

        Returns:
            The handler's Frame result.
        """
        decoded = immutable_params(params)
        return self.handler(inputs, decoded)

    def describe(self) -> RecipeOperationDescription:
        """Describe this operation's inputs, method parameters, and Frame types.

        Parameters come from the decorated method signature after its Frame receiver,
        so they describe the call that Recipe extraction captures. ``Literal``
        annotations also report their accepted values as ``choices``. The receiver
        comes from the imported Frame classes that expose the method, and the result
        type from the method's return annotation.

        Returns:
            An immutable description; ``parameters`` is ``None`` when no method is
                attached to this operation.
        """
        parameters = None
        method_name = None
        receiver: tuple[type[Any], ...] = ()
        returns: type[Any] | Literal["receiver"] | None = None
        if self.method is not None:
            method_name = getattr(self.method, "__name__", type(self.method).__name__)
            signature = inspect.signature(self.method)
            parameters = tuple(
                _describe_parameter(parameter, self.method) for parameter in tuple(signature.parameters.values())[1:]
            )
            receiver = _receiver_classes(self, method_name)
            returns = _returned_frame_class(self.method)
        return RecipeOperationDescription(
            self.operation_id,
            self.version,
            method_name,
            self.binding_patterns,
            parameters,
            receiver,
            returns,
            self.extras,
        )

Attributes

operation_id instance-attribute

version instance-attribute

binding_patterns instance-attribute

handler = field(repr=False) class-attribute instance-attribute

validate_params = field(default=_no_param_validation, repr=False) class-attribute instance-attribute

method = field(default=None, repr=False) class-attribute instance-attribute

extras = () class-attribute instance-attribute

Functions

__post_init__()

Snapshot and validate the complete operation contract.

Source code in wandas/pipeline/registry.py

def __post_init__(self) -> None:
    """Snapshot and validate the complete operation contract."""
    object.__setattr__(self, "binding_patterns", tuple(self.binding_patterns))
    if isinstance(self.extras, str):
        raise TypeError("Recipe operation extras must be a tuple of strings, not a str")
    object.__setattr__(self, "extras", tuple(self.extras))
    if any(not isinstance(extra, str) or not extra.strip() for extra in self.extras):
        raise ValueError("Recipe operation extras must be non-blank strings")
    if not isinstance(self.operation_id, str) or not self.operation_id.strip():
        raise ValueError("Recipe operation id must be a non-blank string")
    if type(self.version) is not int or self.version < 1:
        raise ValueError("Recipe operation version must be a positive integer")
    if not self.binding_patterns:
        raise ValueError("Recipe operation requires at least one binding pattern")
    if any(not pattern for pattern in self.binding_patterns):
        raise ValueError("Recipe binding patterns must each contain at least one input")
    if not all(
        isinstance(pattern, tuple) and all(isinstance(binding, InputBinding) for binding in pattern)
        for pattern in self.binding_patterns
    ):
        raise TypeError("Recipe binding patterns must contain InputBinding tuples")
    if len(set(self.binding_patterns)) != len(self.binding_patterns):
        raise ValueError("Recipe binding patterns must be unique")
    if any(len({binding.role for binding in pattern}) != len(pattern) for pattern in self.binding_patterns):
        raise ValueError("Recipe binding roles must be unique within each pattern")
    kind_patterns = tuple(tuple(binding.kind for binding in pattern) for pattern in self.binding_patterns)
    if len(set(kind_patterns)) != len(kind_patterns):
        raise ValueError("Recipe binding patterns must have unique input kind signatures")
    if not callable(self.handler) or not callable(self.validate_params):
        raise TypeError("Recipe handler and parameter validator must be callable")
    if self.method is not None and not callable(self.method):
        raise TypeError("Recipe operation method must be callable")

accepts(bindings)

Return whether bindings exactly match a declared pattern.

Source code in wandas/pipeline/registry.py

def accepts(self, bindings: tuple[InputBinding, ...]) -> bool:
    """Return whether ``bindings`` exactly match a declared pattern."""
    return bindings in self.binding_patterns

invoke(inputs, params)

Invoke the replay handler after complete-plan validation.

Parameters:

Name	Type	Description	Default
inputs	tuple[Any, ...]	Ordered runtime values matching one declared binding pattern.	required
params	FrozenMap	Canonical parameters already accepted by validate_params.	required

Returns:

Type	Description
Any	The handler's Frame result.

Source code in wandas/pipeline/registry.py

def invoke(self, inputs: tuple[Any, ...], params: FrozenMap) -> Any:
    """Invoke the replay handler after complete-plan validation.

    Args:
        inputs: Ordered runtime values matching one declared binding pattern.
        params: Canonical parameters already accepted by ``validate_params``.

    Returns:
        The handler's Frame result.
    """
    decoded = immutable_params(params)
    return self.handler(inputs, decoded)

describe()

Describe this operation's inputs, method parameters, and Frame types.

Parameters come from the decorated method signature after its Frame receiver, so they describe the call that Recipe extraction captures. Literal annotations also report their accepted values as choices. The receiver comes from the imported Frame classes that expose the method, and the result type from the method's return annotation.

Returns:

Type	Description
RecipeOperationDescription	An immutable description; parameters is None when no method is attached to this operation.

Source code in wandas/pipeline/registry.py

def describe(self) -> RecipeOperationDescription:
    """Describe this operation's inputs, method parameters, and Frame types.

    Parameters come from the decorated method signature after its Frame receiver,
    so they describe the call that Recipe extraction captures. ``Literal``
    annotations also report their accepted values as ``choices``. The receiver
    comes from the imported Frame classes that expose the method, and the result
    type from the method's return annotation.

    Returns:
        An immutable description; ``parameters`` is ``None`` when no method is
            attached to this operation.
    """
    parameters = None
    method_name = None
    receiver: tuple[type[Any], ...] = ()
    returns: type[Any] | Literal["receiver"] | None = None
    if self.method is not None:
        method_name = getattr(self.method, "__name__", type(self.method).__name__)
        signature = inspect.signature(self.method)
        parameters = tuple(
            _describe_parameter(parameter, self.method) for parameter in tuple(signature.parameters.values())[1:]
        )
        receiver = _receiver_classes(self, method_name)
        returns = _returned_frame_class(self.method)
    return RecipeOperationDescription(
        self.operation_id,
        self.version,
        method_name,
        self.binding_patterns,
        parameters,
        receiver,
        returns,
        self.extras,
    )

__init__(operation_id, version, binding_patterns, handler, validate_params=_no_param_validation, method=None, extras=())

RecipeOperationDescription dataclass

Machine-readable description of one registered Recipe operation.

Parameters:

Name	Type	Description	Default
operation_id	str	Stable identifier persisted in Recipe nodes.	required
version	int	Operation contract version.	required
method	str | None	Name of the decorated method, if known. Public calls record operations whose method name is public; contracts kept only to replay older Recipes use private names such as _fft_recipe_v2.	required
binding_patterns	tuple[tuple[InputBinding, ...], ...]	Accepted ordered input-role and input-kind patterns.	required
parameters	tuple[RecipeParameterDescription, ...] | None	Parameters of that method after its Frame receiver, or None when the operation was declared without a method.	required
receiver	tuple[type[Any], ...]	Most general Frame classes the method is called on. Subclasses inherit it unless they override the method with another Recipe declaration, so use :meth:applies_to to check one Frame. Empty when no method is attached or no imported Frame class exposes it.	()
returns	type[Any] | Literal['receiver'] | None	Frame class of the result; "receiver" when the result has the same type as the Frame the method is called on (such as filters and arithmetic); None when the method does not declare a Frame type.	None
extras	tuple[str, ...]	Optional wandas extras the operation needs, using the extra names of pip install "wandas[<extra>]".	()

Source code in wandas/pipeline/registry.py

@dataclass(frozen=True)
class RecipeOperationDescription:
    """Machine-readable description of one registered Recipe operation.

    Args:
        operation_id: Stable identifier persisted in Recipe nodes.
        version: Operation contract version.
        method: Name of the decorated method, if known. Public calls record
            operations whose method name is public; contracts kept only to replay
            older Recipes use private names such as ``_fft_recipe_v2``.
        binding_patterns: Accepted ordered input-role and input-kind patterns.
        parameters: Parameters of that method after its Frame receiver, or ``None``
            when the operation was declared without a method.
        receiver: Most general Frame classes the method is called on. Subclasses
            inherit it unless they override the method with another Recipe
            declaration, so use :meth:`applies_to` to check one Frame. Empty when no
            method is attached or no imported Frame class exposes it.
        returns: Frame class of the result; ``"receiver"`` when the result has the
            same type as the Frame the method is called on (such as filters and
            arithmetic); ``None`` when the method does not declare a Frame type.
        extras: Optional ``wandas`` extras the operation needs, using the extra
            names of ``pip install "wandas[<extra>]"``.
    """

    operation_id: str
    version: int
    method: str | None
    binding_patterns: tuple[tuple[InputBinding, ...], ...]
    parameters: tuple[RecipeParameterDescription, ...] | None
    receiver: tuple[type[Any], ...] = ()
    returns: type[Any] | Literal["receiver"] | None = None
    extras: tuple[str, ...] = ()

    def applies_to(self, frame: Any) -> bool:
        """Return whether calling ``method`` on ``frame`` records this operation.

        Args:
            frame: A Frame or a Frame class.

        Returns:
            ``False`` also for a subclass of ``receiver`` that overrides the method
            with another Recipe declaration.
        """
        cls = frame if isinstance(frame, type) else type(frame)
        if self.method is None or not self.receiver or not issubclass(cls, self.receiver):
            return False
        operation = getattr(getattr(cls, self.method, None), "__wandas_recipe_operation__", None)
        return operation is not None and (operation.operation_id, operation.version) == (
            self.operation_id,
            self.version,
        )

Attributes

operation_id instance-attribute

version instance-attribute

method instance-attribute

binding_patterns instance-attribute

parameters instance-attribute

receiver = () class-attribute instance-attribute

returns = None class-attribute instance-attribute

extras = () class-attribute instance-attribute

Functions

applies_to(frame)

Return whether calling method on frame records this operation.

Parameters:

Name	Type	Description	Default
frame	Any	A Frame or a Frame class.	required

Returns:

Type	Description
bool	False also for a subclass of receiver that overrides the method
bool	with another Recipe declaration.

Source code in wandas/pipeline/registry.py

def applies_to(self, frame: Any) -> bool:
    """Return whether calling ``method`` on ``frame`` records this operation.

    Args:
        frame: A Frame or a Frame class.

    Returns:
        ``False`` also for a subclass of ``receiver`` that overrides the method
        with another Recipe declaration.
    """
    cls = frame if isinstance(frame, type) else type(frame)
    if self.method is None or not self.receiver or not issubclass(cls, self.receiver):
        return False
    operation = getattr(getattr(cls, self.method, None), "__wandas_recipe_operation__", None)
    return operation is not None and (operation.operation_id, operation.version) == (
        self.operation_id,
        self.version,
    )

__init__(operation_id, version, method, binding_patterns, parameters, receiver=(), returns=None, extras=())

RecipeParameterDescription dataclass

Machine-readable description of one operation method parameter.

Parameters:

Name	Type	Description	Default
name	str	Parameter name accepted by the public method.	required
kind	str	:class:inspect.Parameter kind name, such as "KEYWORD_ONLY".	required
required	bool	Whether the call must supply a value.	required
default	Any	Deep copy of the default Python value, or None when required is true. A default that cannot be deep-copied, such as a lock, is returned as the same object.	required
annotation	str | None	Annotation text as written for postponed annotations, or the :func:inspect.formatannotation text of an evaluated annotation; None when the parameter is unannotated.	required
choices	tuple[Any, ...] | None	Accepted values when the annotation is a Literal (optionally joined with None); otherwise None.	required

Source code in wandas/pipeline/registry.py

@dataclass(frozen=True)
class RecipeParameterDescription:
    """Machine-readable description of one operation method parameter.

    Args:
        name: Parameter name accepted by the public method.
        kind: :class:`inspect.Parameter` kind name, such as ``"KEYWORD_ONLY"``.
        required: Whether the call must supply a value.
        default: Deep copy of the default Python value, or ``None`` when ``required``
            is true. A default that cannot be deep-copied, such as a lock, is
            returned as the same object.
        annotation: Annotation text as written for postponed annotations, or the
            :func:`inspect.formatannotation` text of an evaluated annotation; ``None``
            when the parameter is unannotated.
        choices: Accepted values when the annotation is a ``Literal`` (optionally
            joined with ``None``); otherwise ``None``.
    """

    name: str
    kind: str
    required: bool
    default: Any
    annotation: str | None
    choices: tuple[Any, ...] | None

Attributes

name instance-attribute

kind instance-attribute

required instance-attribute

default instance-attribute

annotation instance-attribute

choices instance-attribute

Functions

__init__(name, kind, required, default, annotation, choices)

RecipeRegistry dataclass

Immutable collection of versioned Recipe operation contracts.

Parameters:

Name	Type	Description	Default
operations	Iterable[RecipeOperation]	Initial operation definitions. Each (operation_id, version) pair must be unique.	()

Raises:

Type	Description
TypeError	If an entry is not a :class:RecipeOperation.
ValueError	If two entries use the same identifier and version.

Source code in wandas/pipeline/registry.py

@dataclass(frozen=True)
class RecipeRegistry:
    """Immutable collection of versioned Recipe operation contracts.

    Args:
        operations: Initial operation definitions. Each ``(operation_id, version)``
            pair must be unique.

    Raises:
        TypeError: If an entry is not a :class:`RecipeOperation`.
        ValueError: If two entries use the same identifier and version.
    """

    operations: tuple[RecipeOperation, ...] = ()
    _by_key: Mapping[tuple[str, int], RecipeOperation] = field(init=False, repr=False, compare=False)

    def __init__(self, operations: Iterable[RecipeOperation] = ()) -> None:
        """Snapshot definitions and build a read-only lookup index."""
        normalized = tuple(operations)
        by_key: dict[tuple[str, int], RecipeOperation] = {}
        for operation in normalized:
            if not isinstance(operation, RecipeOperation):
                raise TypeError("RecipeRegistry entries must be RecipeOperation values")
            key = (operation.operation_id, operation.version)
            if key in by_key:
                raise ValueError(f"Recipe operation is already registered: {key!r}")
            by_key[key] = operation
        object.__setattr__(self, "operations", normalized)
        object.__setattr__(self, "_by_key", MappingProxyType(by_key))

    def with_operation(self, operation: RecipeOperation) -> RecipeRegistry:
        """Return a new registry containing one additional operation.

        Args:
            operation: Definition to append without changing this registry.

        Returns:
            A new immutable registry.

        Raises:
            ValueError: If the operation identifier and version already exist.
            TypeError: If ``operation`` is not a :class:`RecipeOperation`.
        """
        return RecipeRegistry((*self.operations, operation))

    def require(self, operation_id: str, version: int) -> RecipeOperation:
        """Return an exact registered operation definition.

        Args:
            operation_id: Stable operation identifier.
            version: Required contract version.

        Returns:
            The matching operation definition.

        Raises:
            KeyError: If no exact identifier-version pair is registered.
        """
        try:
            return self._by_key[(operation_id, version)]
        except KeyError as exc:
            raise KeyError(f"Recipe operation is not registered: {operation_id!r} version {version}") from exc

    def describe(self) -> tuple[RecipeOperationDescription, ...]:
        """Describe every registered operation in registration order.

        Use this catalog to build parameter forms, documentation, or tool schemas
        from the same declarations that Recipe validation and replay use. Released
        contracts kept only to replay older Recipes are included; their ``method``
        is private (one leading underscore, such as ``_fft_recipe_v2``). Public
        calls record the operations whose ``method`` is public, including operator
        methods such as ``__add__``.

        Returns:
            One :class:`RecipeOperationDescription` per registered operation.
        """
        return tuple(operation.describe() for operation in self.operations)

Attributes

operations = () class-attribute instance-attribute

Functions

__init__(operations=())

Snapshot definitions and build a read-only lookup index.

Source code in wandas/pipeline/registry.py

def __init__(self, operations: Iterable[RecipeOperation] = ()) -> None:
    """Snapshot definitions and build a read-only lookup index."""
    normalized = tuple(operations)
    by_key: dict[tuple[str, int], RecipeOperation] = {}
    for operation in normalized:
        if not isinstance(operation, RecipeOperation):
            raise TypeError("RecipeRegistry entries must be RecipeOperation values")
        key = (operation.operation_id, operation.version)
        if key in by_key:
            raise ValueError(f"Recipe operation is already registered: {key!r}")
        by_key[key] = operation
    object.__setattr__(self, "operations", normalized)
    object.__setattr__(self, "_by_key", MappingProxyType(by_key))

with_operation(operation)

Return a new registry containing one additional operation.

Parameters:

Name	Type	Description	Default
operation	RecipeOperation	Definition to append without changing this registry.	required

Returns:

Type	Description
RecipeRegistry	A new immutable registry.

Raises:

Type	Description
ValueError	If the operation identifier and version already exist.
TypeError	If operation is not a :class:RecipeOperation.

Source code in wandas/pipeline/registry.py

def with_operation(self, operation: RecipeOperation) -> RecipeRegistry:
    """Return a new registry containing one additional operation.

    Args:
        operation: Definition to append without changing this registry.

    Returns:
        A new immutable registry.

    Raises:
        ValueError: If the operation identifier and version already exist.
        TypeError: If ``operation`` is not a :class:`RecipeOperation`.
    """
    return RecipeRegistry((*self.operations, operation))

require(operation_id, version)

Return an exact registered operation definition.

Parameters:

Name	Type	Description	Default
operation_id	str	Stable operation identifier.	required
version	int	Required contract version.	required

Returns:

Type	Description
RecipeOperation	The matching operation definition.

Raises:

Type	Description
KeyError	If no exact identifier-version pair is registered.

Source code in wandas/pipeline/registry.py

def require(self, operation_id: str, version: int) -> RecipeOperation:
    """Return an exact registered operation definition.

    Args:
        operation_id: Stable operation identifier.
        version: Required contract version.

    Returns:
        The matching operation definition.

    Raises:
        KeyError: If no exact identifier-version pair is registered.
    """
    try:
        return self._by_key[(operation_id, version)]
    except KeyError as exc:
        raise KeyError(f"Recipe operation is not registered: {operation_id!r} version {version}") from exc

describe()

Describe every registered operation in registration order.

Use this catalog to build parameter forms, documentation, or tool schemas from the same declarations that Recipe validation and replay use. Released contracts kept only to replay older Recipes are included; their method is private (one leading underscore, such as _fft_recipe_v2). Public calls record the operations whose method is public, including operator methods such as __add__.

Returns:

Name	Type	Description
One	tuple[RecipeOperationDescription, ...]	class:RecipeOperationDescription per registered operation.

Source code in wandas/pipeline/registry.py

def describe(self) -> tuple[RecipeOperationDescription, ...]:
    """Describe every registered operation in registration order.

    Use this catalog to build parameter forms, documentation, or tool schemas
    from the same declarations that Recipe validation and replay use. Released
    contracts kept only to replay older Recipes are included; their ``method``
    is private (one leading underscore, such as ``_fft_recipe_v2``). Public
    calls record the operations whose ``method`` is public, including operator
    methods such as ``__add__``.

    Returns:
        One :class:`RecipeOperationDescription` per registered operation.
    """
    return tuple(operation.describe() for operation in self.operations)

Functions

recipe_definition(value)

Return the operation declaration attached by :func:recipe_operation.

Parameters:

Name	Type	Description	Default
value	object	Decorated public method or an equivalent statically inspected member.	required

Returns:

Type	Description
RecipeOperation	The immutable operation contract ready to add to a :class:RecipeRegistry.

Raises:

Type	Description
TypeError	If value has no Recipe operation declaration.

Source code in wandas/pipeline/decorators.py

def recipe_definition(value: object) -> RecipeOperation:
    """Return the operation declaration attached by :func:`recipe_operation`.

    Args:
        value: Decorated public method or an equivalent statically inspected member.

    Returns:
        The immutable operation contract ready to add to a :class:`RecipeRegistry`.

    Raises:
        TypeError: If ``value`` has no Recipe operation declaration.
    """
    definition = getattr(value, "__wandas_recipe_operation__", None)
    if not isinstance(definition, RecipeOperation):
        raise TypeError("Object has no Recipe operation declaration")
    return definition

recipe_operation(operation_id, *, version=1, bindings=None, binding_patterns=None, capture=None, handler=None, validate_params=None, extras=())

Declare and capture one portable public Frame operation.

The returned decorator attaches a :class:RecipeOperation declaration to the wrapped method and records one authoritative semantic lineage node per public call. The default unary replay handler delegates to the wrapped public method. An extension that supplies an explicit handler is responsible for keeping that handler behaviorally equivalent to its public method.

Parameters:

Name	Type	Description	Default
operation_id	str	Stable serialized identifier for the public operation.	required
version	int	Positive contract version for operation_id.	1
bindings	tuple[InputBinding, ...] | None	Ordered input contract for one invocation shape. Defaults to one Frame binding named frame.	None
binding_patterns	tuple[tuple[InputBinding, ...], ...] | None	Alternative accepted input contracts for operations that support more than one shape, such as Frame-or-array operands. Mutually exclusive with bindings.	None
capture	CaptureResolver | None	Callback that derives actual bindings, lineage parents, and portable parameters from a public call. Required for non-unary contracts.	None
handler	RecipeHandler | None	Replay callback receiving ordered runtime inputs and an immutable parameter mapping. Required for non-unary contracts; unary operations use the wrapped method by default.	None
validate_params	ParamValidator | None	Optional validator called on immutable decoded parameters during complete-plan validation.	None
extras	tuple[str, ...]	Optional wandas extras the operation needs, such as ("effects",); :meth:RecipeRegistry.describe reports them.	()

Returns:

Type	Description
Callable[[Callable[P, R]], Callable[P, R]]	A method decorator that preserves the wrapped signature and exposes its Recipe declaration through :func:recipe_definition.

Raises:

Type	Description
TypeError	If capture is supplied but is not callable.
ValueError	If declarations conflict, binding patterns are empty, a required capture or handler is missing, or the method signature is unsupported.

Source code in wandas/pipeline/decorators.py

def recipe_operation(
    operation_id: str,
    *,
    version: int = 1,
    bindings: tuple[InputBinding, ...] | None = None,
    binding_patterns: tuple[tuple[InputBinding, ...], ...] | None = None,
    capture: CaptureResolver | None = None,
    handler: RecipeHandler | None = None,
    validate_params: ParamValidator | None = None,
    extras: tuple[str, ...] = (),
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Declare and capture one portable public Frame operation.

    The returned decorator attaches a :class:`RecipeOperation` declaration to the
    wrapped method and records one authoritative semantic lineage node per public call.
    The default unary replay handler delegates to the wrapped public method. An
    extension that supplies an explicit handler is responsible for keeping that handler
    behaviorally equivalent to its public method.

    Args:
        operation_id: Stable serialized identifier for the public operation.
        version: Positive contract version for ``operation_id``.
        bindings: Ordered input contract for one invocation shape. Defaults to one
            Frame binding named ``frame``.
        binding_patterns: Alternative accepted input contracts for operations that
            support more than one shape, such as Frame-or-array operands. Mutually
            exclusive with ``bindings``.
        capture: Callback that derives actual bindings, lineage parents, and portable
            parameters from a public call. Required for non-unary contracts.
        handler: Replay callback receiving ordered runtime inputs and an immutable
            parameter mapping. Required for non-unary contracts; unary operations use
            the wrapped method by default.
        validate_params: Optional validator called on immutable decoded parameters
            during complete-plan validation.
        extras: Optional ``wandas`` extras the operation needs, such as
            ``("effects",)``; :meth:`RecipeRegistry.describe` reports them.

    Returns:
        A method decorator that preserves the wrapped signature and exposes its
            Recipe declaration through :func:`recipe_definition`.

    Raises:
        TypeError: If ``capture`` is supplied but is not callable.
        ValueError: If declarations conflict, binding patterns are empty, a required
            capture or handler is missing, or the method signature is unsupported.
    """
    if bindings is not None and binding_patterns is not None:
        raise ValueError("Specify either bindings or binding_patterns, not both")
    if capture is not None and not callable(capture):
        raise TypeError("Recipe capture must be callable")
    declared_bindings = bindings if bindings is not None else (InputBinding("frame", "frame"),)
    patterns = binding_patterns if binding_patterns is not None else (declared_bindings,)
    if not patterns:
        raise ValueError("Recipe operation requires at least one binding pattern")
    unary_frame_contract = (
        len(patterns) == 1
        and isinstance(patterns[0], tuple)
        and len(patterns[0]) == 1
        and isinstance(patterns[0][0], InputBinding)
        and patterns[0][0].kind == "frame"
    )
    if handler is None and not unary_frame_contract:
        raise ValueError("Recipe operations with non-unary Frame bindings require an explicit handler")
    if capture is None and not unary_frame_contract:
        raise ValueError("Recipe operations with non-unary Frame bindings require an explicit capture")

    def decorate(method: Callable[P, R]) -> Callable[P, R]:
        """Attach semantic capture and a registry-ready operation definition."""
        signature = inspect.signature(method)
        parameters = tuple(signature.parameters.values())
        if not parameters or parameters[0].kind not in {
            inspect.Parameter.POSITIONAL_ONLY,
            inspect.Parameter.POSITIONAL_OR_KEYWORD,
        }:
            raise ValueError("Recipe operation methods require a positional Frame receiver")
        method_parameters = parameters[1:]
        if handler is None and any(
            parameter.kind in {inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.VAR_POSITIONAL}
            for parameter in method_parameters
        ):
            raise ValueError("Default Recipe handlers do not support positional-only or variadic positional parameters")
        capture_resolver = capture if capture is not None else _unary_capture(patterns[0][0])

        @wraps(method)
        def semantic_call(*args: P.args, **kwargs: P.kwargs) -> R:
            """Capture a public call once, then execute it under that lineage."""
            if active_semantic_lineage() is not None:
                return method(*args, **kwargs)
            params = _bound_arguments(signature, cast(tuple[Any, ...], args), kwargs)
            captured = capture_resolver(cast(tuple[Any, ...], args), params)
            try:
                frozen = freeze_params(captured.params)
                recipe_error = captured.recipe_error
            except (TypeError, ValueError) as exc:
                frozen = _freeze_display_params(captured.params)
                recipe_error = captured.recipe_error or f"Public arguments are not portable Recipe values: {exc}"
            operation = __operation_for_capture(captured.bindings, frozen)
            lineage = LineageNode(operation, captured.parents, recipe_error=recipe_error)
            return cast(R, invoke_semantic(method, lineage, *args, **kwargs))

        def default_handler(inputs: tuple[Any, ...], params: Mapping[str, Any]) -> Any:
            """Replay a unary operation through its decorated public method."""
            return semantic_call(inputs[0], **dict(params))

        definition = RecipeOperation(
            operation_id,
            version,
            patterns,
            handler if handler is not None else default_handler,
            validate_params if validate_params is not None else (lambda _params: None),
            method=method,
            extras=extras,
        )

        def __operation_for_capture(actual_bindings: tuple[InputBinding, ...], params: Any) -> Any:
            """Create semantic intent after checking captured binding agreement."""
            from wandas.processing.semantic import SemanticOperation

            if not definition.accepts(actual_bindings):
                raise RuntimeError(
                    f"Public Recipe capture produced undeclared bindings for {operation_id!r}: {actual_bindings!r}"
                )
            return SemanticOperation(operation_id, version, actual_bindings, params)

        setattr(semantic_call, "__wandas_recipe_operation__", definition)
        return semantic_call

    return decorate

default_recipe_registry() cached

Return the cached immutable registry of built-in Frame operations.

Returns:

Type	Description
RecipeRegistry	A process-wide immutable registry. Use :meth:RecipeRegistry.with_operation to derive an extension registry without mutating the built-in value.

Source code in wandas/pipeline/registry.py

@cache
def default_recipe_registry() -> RecipeRegistry:
    """Return the cached immutable registry of built-in Frame operations.

    Returns:
        A process-wide immutable registry. Use :meth:`RecipeRegistry.with_operation`
            to derive an extension registry without mutating the built-in value.
    """
    from wandas.pipeline.builtins import builtin_recipe_operations

    return RecipeRegistry(builtin_recipe_operations())

wandas.pipeline.sklearn

Scikit-learn-compatible adapters for declared Wandas Frame operations.

Attributes

__all__ = ['BandPassFilter', 'HighPassFilter', 'LowPassFilter', 'Normalize', 'RemoveDC', 'WandasOperationTransformer'] module-attribute

Classes

BaseEstimator

Import-time placeholder used only to define unavailable adapters.

Source code in wandas/pipeline/sklearn.py

class BaseEstimator:  # type: ignore[no-redef]
    """Import-time placeholder used only to define unavailable adapters."""

    pass

TransformerMixin

Import-time placeholder used only to define unavailable adapters.

Source code in wandas/pipeline/sklearn.py

class TransformerMixin:  # type: ignore[no-redef]
    """Import-time placeholder used only to define unavailable adapters."""

    pass

WandasOperationTransformer

Bases: TransformerMixin, BaseEstimator

Stateless sklearn transformer for a declared Wandas Frame operation.

Parameters:

Name	Type	Description	Default
operation	str	Public Frame method name decorated with @recipe_operation.	required
**params	Any	Keyword arguments forwarded to that method by :meth:transform.	{}

Raises:

Type	Description
ImportError	If scikit-learn is not installed.

Source code in wandas/pipeline/sklearn.py

class WandasOperationTransformer(TransformerMixin, BaseEstimator):  # type: ignore[misc]
    """Stateless sklearn transformer for a declared Wandas Frame operation.

    Args:
        operation: Public Frame method name decorated with ``@recipe_operation``.
        **params: Keyword arguments forwarded to that method by :meth:`transform`.

    Raises:
        ImportError: If scikit-learn is not installed.
    """

    _param_names: tuple[str, ...] | None = None

    def __init__(self, operation: str, **params: Any) -> None:
        """Store operation identity and estimator parameters without fitting state."""
        _require_sklearn()
        self.operation = operation
        if self._param_names is None:
            self._params = dict(params)

    def _resolved_params(self) -> dict[str, Any]:
        """Return current estimator parameters for the wrapped Frame call."""
        if self._param_names is None:
            return dict(self._params)
        return {name: getattr(self, name) for name in self._param_names}

    def fit(self, X: Any, y: Any = None) -> WandasOperationTransformer:  # noqa: N803
        """Return this stateless transformer without inspecting training data.

        Args:
            X: Accepted for sklearn estimator compatibility and left untouched.
            y: Optional target accepted for sklearn estimator compatibility.

        Returns:
            This already-fitted stateless transformer.
        """
        return self

    def __sklearn_is_fitted__(self) -> bool:
        """Report that this stateless transformer requires no fitted attributes."""
        return True

    def transform(self, X: Any) -> Any:  # noqa: N803
        """Apply the declared public operation to a Wandas Frame.

        Args:
            X: Frame exposing the configured declared public operation.

        Returns:
            The Frame result returned by that operation.

        Raises:
            ValueError: If ``operation`` does not name a declared Recipe-capable
                public method on ``X``.
        """
        declaration = inspect.getattr_static(X, self.operation, None)
        try:
            recipe_definition(declaration)
        except TypeError as exc:
            raise ValueError(
                f"operation must name a declared public Recipe Frame method, got {self.operation!r}"
            ) from exc
        method = getattr(X, self.operation)
        return method(**self._resolved_params())

    def get_params(self, deep: bool = True) -> dict[str, Any]:
        """Return estimator parameters using sklearn's cloning contract.

        Args:
            deep: Accepted for sklearn compatibility; this adapter has no nested
                estimators.

        Returns:
            A fresh parameter dictionary.
        """
        params = self._resolved_params()
        if self._param_names is None:
            return {"operation": self.operation, **params}
        return params

    def set_params(self, **params: Any) -> WandasOperationTransformer:
        """Update estimator parameters in place using sklearn conventions.

        Only the given parameters change. Passing a different ``operation`` to the
        generic adapter replaces its parameters with the given ones.

        Args:
            **params: Parameter values to update.

        Returns:
            This transformer.

        Raises:
            TypeError: If a generic ``operation`` update is not a string.
            ValueError: If a parameter is invalid for the current adapter configuration.
        """
        if self._param_names is None and "operation" in params:
            operation = params.pop("operation")
            if not isinstance(operation, str):
                raise TypeError(f"operation must be a string, got {type(operation).__name__}")
            if operation == self.operation:
                self._params.update(params)
            else:
                # Parameters of the previous method do not apply to another one.
                self.operation = operation
                self._params = dict(params)
            return self
        valid_params = self.get_params(deep=True)
        for key, value in params.items():
            if key not in valid_params:
                valid_names = ", ".join(sorted(valid_params))
                raise ValueError(f"Invalid parameter {key!r}. Valid parameters are: {valid_names}")
            if self._param_names is None:
                self._params[key] = value
            else:
                setattr(self, key, value)
        return self

Attributes

operation = operation instance-attribute

Functions

__init__(operation, **params)

Store operation identity and estimator parameters without fitting state.

Source code in wandas/pipeline/sklearn.py

def __init__(self, operation: str, **params: Any) -> None:
    """Store operation identity and estimator parameters without fitting state."""
    _require_sklearn()
    self.operation = operation
    if self._param_names is None:
        self._params = dict(params)

fit(X, y=None)

Return this stateless transformer without inspecting training data.

Parameters:

Name	Type	Description	Default
X	Any	Accepted for sklearn estimator compatibility and left untouched.	required
y	Any	Optional target accepted for sklearn estimator compatibility.	None

Returns:

Type	Description
WandasOperationTransformer	This already-fitted stateless transformer.

Source code in wandas/pipeline/sklearn.py

def fit(self, X: Any, y: Any = None) -> WandasOperationTransformer:  # noqa: N803
    """Return this stateless transformer without inspecting training data.

    Args:
        X: Accepted for sklearn estimator compatibility and left untouched.
        y: Optional target accepted for sklearn estimator compatibility.

    Returns:
        This already-fitted stateless transformer.
    """
    return self

__sklearn_is_fitted__()

Report that this stateless transformer requires no fitted attributes.

Source code in wandas/pipeline/sklearn.py

def __sklearn_is_fitted__(self) -> bool:
    """Report that this stateless transformer requires no fitted attributes."""
    return True

transform(X)

Apply the declared public operation to a Wandas Frame.

Parameters:

Name	Type	Description	Default
X	Any	Frame exposing the configured declared public operation.	required

Returns:

Type	Description
Any	The Frame result returned by that operation.

Raises:

Type	Description
ValueError	If operation does not name a declared Recipe-capable public method on X.

Source code in wandas/pipeline/sklearn.py

def transform(self, X: Any) -> Any:  # noqa: N803
    """Apply the declared public operation to a Wandas Frame.

    Args:
        X: Frame exposing the configured declared public operation.

    Returns:
        The Frame result returned by that operation.

    Raises:
        ValueError: If ``operation`` does not name a declared Recipe-capable
            public method on ``X``.
    """
    declaration = inspect.getattr_static(X, self.operation, None)
    try:
        recipe_definition(declaration)
    except TypeError as exc:
        raise ValueError(
            f"operation must name a declared public Recipe Frame method, got {self.operation!r}"
        ) from exc
    method = getattr(X, self.operation)
    return method(**self._resolved_params())

get_params(deep=True)

Return estimator parameters using sklearn's cloning contract.

Parameters:

Name	Type	Description	Default
deep	bool	Accepted for sklearn compatibility; this adapter has no nested estimators.	True

Returns:

Type	Description
dict[str, Any]	A fresh parameter dictionary.

Source code in wandas/pipeline/sklearn.py

def get_params(self, deep: bool = True) -> dict[str, Any]:
    """Return estimator parameters using sklearn's cloning contract.

    Args:
        deep: Accepted for sklearn compatibility; this adapter has no nested
            estimators.

    Returns:
        A fresh parameter dictionary.
    """
    params = self._resolved_params()
    if self._param_names is None:
        return {"operation": self.operation, **params}
    return params

set_params(**params)

Update estimator parameters in place using sklearn conventions.

Only the given parameters change. Passing a different operation to the generic adapter replaces its parameters with the given ones.

Parameters:

Name	Type	Description	Default
**params	Any	Parameter values to update.	{}

Returns:

Type	Description
WandasOperationTransformer	This transformer.

Raises:

Type	Description
TypeError	If a generic operation update is not a string.
ValueError	If a parameter is invalid for the current adapter configuration.

Source code in wandas/pipeline/sklearn.py

def set_params(self, **params: Any) -> WandasOperationTransformer:
    """Update estimator parameters in place using sklearn conventions.

    Only the given parameters change. Passing a different ``operation`` to the
    generic adapter replaces its parameters with the given ones.

    Args:
        **params: Parameter values to update.

    Returns:
        This transformer.

    Raises:
        TypeError: If a generic ``operation`` update is not a string.
        ValueError: If a parameter is invalid for the current adapter configuration.
    """
    if self._param_names is None and "operation" in params:
        operation = params.pop("operation")
        if not isinstance(operation, str):
            raise TypeError(f"operation must be a string, got {type(operation).__name__}")
        if operation == self.operation:
            self._params.update(params)
        else:
            # Parameters of the previous method do not apply to another one.
            self.operation = operation
            self._params = dict(params)
        return self
    valid_params = self.get_params(deep=True)
    for key, value in params.items():
        if key not in valid_params:
            valid_names = ", ".join(sorted(valid_params))
            raise ValueError(f"Invalid parameter {key!r}. Valid parameters are: {valid_names}")
        if self._param_names is None:
            self._params[key] = value
        else:
            setattr(self, key, value)
    return self

HighPassFilter

Bases: WandasOperationTransformer

Stateless sklearn adapter for Frame.high_pass_filter.

Parameters:

Name	Type	Description	Default
cutoff	float	High-pass cutoff frequency in hertz.	required
order	int	Filter order.	4

Source code in wandas/pipeline/sklearn.py

class HighPassFilter(WandasOperationTransformer):
    """Stateless sklearn adapter for ``Frame.high_pass_filter``.

    Args:
        cutoff: High-pass cutoff frequency in hertz.
        order: Filter order.
    """

    _param_names = ("cutoff", "order")

    def __init__(self, cutoff: float, order: int = 4) -> None:
        """Initialize high-pass filter parameters."""
        self.cutoff = cutoff
        self.order = order
        super().__init__("high_pass_filter")

Attributes

cutoff = cutoff instance-attribute

order = order instance-attribute

Functions

__init__(cutoff, order=4)

Initialize high-pass filter parameters.

Source code in wandas/pipeline/sklearn.py

def __init__(self, cutoff: float, order: int = 4) -> None:
    """Initialize high-pass filter parameters."""
    self.cutoff = cutoff
    self.order = order
    super().__init__("high_pass_filter")

LowPassFilter

Bases: WandasOperationTransformer

Stateless sklearn adapter for Frame.low_pass_filter.

Parameters:

Name	Type	Description	Default
cutoff	float	Low-pass cutoff frequency in hertz.	required
order	int	Filter order.	4

Source code in wandas/pipeline/sklearn.py

class LowPassFilter(WandasOperationTransformer):
    """Stateless sklearn adapter for ``Frame.low_pass_filter``.

    Args:
        cutoff: Low-pass cutoff frequency in hertz.
        order: Filter order.
    """

    _param_names = ("cutoff", "order")

    def __init__(self, cutoff: float, order: int = 4) -> None:
        """Initialize low-pass filter parameters."""
        self.cutoff = cutoff
        self.order = order
        super().__init__("low_pass_filter")

Attributes

cutoff = cutoff instance-attribute

order = order instance-attribute

Functions

__init__(cutoff, order=4)

Initialize low-pass filter parameters.

Source code in wandas/pipeline/sklearn.py

def __init__(self, cutoff: float, order: int = 4) -> None:
    """Initialize low-pass filter parameters."""
    self.cutoff = cutoff
    self.order = order
    super().__init__("low_pass_filter")

BandPassFilter

Bases: WandasOperationTransformer

Stateless sklearn adapter for Frame.band_pass_filter.

Parameters:

Name	Type	Description	Default
low_cutoff	float	Lower cutoff frequency in hertz.	required
high_cutoff	float	Upper cutoff frequency in hertz.	required
order	int	Filter order.	4

Source code in wandas/pipeline/sklearn.py

class BandPassFilter(WandasOperationTransformer):
    """Stateless sklearn adapter for ``Frame.band_pass_filter``.

    Args:
        low_cutoff: Lower cutoff frequency in hertz.
        high_cutoff: Upper cutoff frequency in hertz.
        order: Filter order.
    """

    _param_names = ("low_cutoff", "high_cutoff", "order")

    def __init__(self, low_cutoff: float, high_cutoff: float, order: int = 4) -> None:
        """Initialize band-pass filter parameters."""
        self.low_cutoff = low_cutoff
        self.high_cutoff = high_cutoff
        self.order = order
        super().__init__("band_pass_filter")

Attributes

low_cutoff = low_cutoff instance-attribute

high_cutoff = high_cutoff instance-attribute

order = order instance-attribute

Functions

__init__(low_cutoff, high_cutoff, order=4)

Initialize band-pass filter parameters.

Source code in wandas/pipeline/sklearn.py

def __init__(self, low_cutoff: float, high_cutoff: float, order: int = 4) -> None:
    """Initialize band-pass filter parameters."""
    self.low_cutoff = low_cutoff
    self.high_cutoff = high_cutoff
    self.order = order
    super().__init__("band_pass_filter")

Normalize

Bases: WandasOperationTransformer

Stateless sklearn adapter for Frame.normalize.

Parameters:

Name	Type	Description	Default
norm	float | None	Norm used for amplitude normalization.	inf
axis	int | None	Axis normalized independently, or None for global normalization.	-1
threshold	float | None	Optional magnitude threshold treated as zero.	None
fill	bool | None	Optional zero-norm fill behavior forwarded to the Frame operation.	None

Source code in wandas/pipeline/sklearn.py

class Normalize(WandasOperationTransformer):
    """Stateless sklearn adapter for ``Frame.normalize``.

    Args:
        norm: Norm used for amplitude normalization.
        axis: Axis normalized independently, or ``None`` for global normalization.
        threshold: Optional magnitude threshold treated as zero.
        fill: Optional zero-norm fill behavior forwarded to the Frame operation.
    """

    _param_names = ("norm", "axis", "threshold", "fill")

    def __init__(
        self,
        norm: float | None = np.inf,
        axis: int | None = -1,
        threshold: float | None = None,
        fill: bool | None = None,
    ) -> None:
        """Initialize normalization parameters."""
        self.norm = norm
        self.axis = axis
        self.threshold = threshold
        self.fill = fill
        super().__init__("normalize")

Attributes

norm = norm instance-attribute

axis = axis instance-attribute

threshold = threshold instance-attribute

fill = fill instance-attribute

Functions

__init__(norm=np.inf, axis=-1, threshold=None, fill=None)

Initialize normalization parameters.

Source code in wandas/pipeline/sklearn.py

def __init__(
    self,
    norm: float | None = np.inf,
    axis: int | None = -1,
    threshold: float | None = None,
    fill: bool | None = None,
) -> None:
    """Initialize normalization parameters."""
    self.norm = norm
    self.axis = axis
    self.threshold = threshold
    self.fill = fill
    super().__init__("normalize")

RemoveDC

Bases: WandasOperationTransformer

Stateless sklearn adapter for Frame.remove_dc.

Source code in wandas/pipeline/sklearn.py

class RemoveDC(WandasOperationTransformer):
    """Stateless sklearn adapter for ``Frame.remove_dc``."""

    _param_names: tuple[str, ...] = ()

    def __init__(self) -> None:
        """Initialize the parameter-free DC-removal transformer."""
        super().__init__("remove_dc")

Functions

__init__()

Initialize the parameter-free DC-removal transformer.

Source code in wandas/pipeline/sklearn.py

def __init__(self) -> None:
    """Initialize the parameter-free DC-removal transformer."""
    super().__init__("remove_dc")

Functions