Pipeline Recipes API

wandas.pipeline provides the portable Recipe surface. Most users need only RecipePlan; registry and decorator APIs are for extensions with stable public Frame methods.

Start with the Recipe tutorial for a complete workflow or the Recipe how-to for concise task recipes.

RecipePlan

Create plans with RecipePlan.from_frame(), RecipePlan.from_dict(), or RecipePlan.load(). Persist a standalone strict JSON artifact with RecipePlan.save(). The direct graph constructor is internal and is not a supported compatibility surface.

wandas.pipeline.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.

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
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.

    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.
        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

    def __init__(
        self,
        inputs: Iterable[RecipeInput],
        nodes: Iterable[RecipeNode],
        output: 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)
        validate_recipe_plan(self, registry=registry)

    @classmethod
    def from_frame(
        cls,
        frame: Any,
        *,
        input_names: tuple[str, ...] | 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.
            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, or ``input_names`` has the wrong length.
        """
        from wandas.pipeline.compiler import LineageRecipeCompiler

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

    @classmethod
    def from_dict(
        cls,
        payload: Mapping[str, Any],
        *,
        registry: RecipeRegistry | None = None,
    ) -> RecipePlan:
        """Load and validate a Recipe schema-2 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 2.
        """
        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],
        *,
        registry: RecipeRegistry | None = None,
    ) -> Any:
        """Build the Recipe output 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.
            registry: Registry used for validation and operation execution. Supply
                the same extension registry used during extraction and loading.

        Returns:
            The output Frame 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, an
                operation fails, or an operation violates the Frame lineage contract.
            RecipeValidationError: If the plan is invalid for ``registry``.
        """
        return RecipeExecutor(registry=registry).execute(self, inputs)

Attributes

inputs instance-attribute

nodes instance-attribute

output instance-attribute

Functions

__init__(inputs, nodes, output, *, 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,
    *,
    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)
    validate_recipe_plan(self, registry=registry)

from_frame(frame, *, input_names=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
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, or input_names has the wrong length.

Source code in wandas/pipeline/model.py

@classmethod
def from_frame(
    cls,
    frame: Any,
    *,
    input_names: tuple[str, ...] | 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.
        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, or ``input_names`` has the wrong length.
    """
    from wandas.pipeline.compiler import LineageRecipeCompiler

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

from_dict(payload, *, registry=None) classmethod

Load and validate a Recipe schema-2 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 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 2.

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 2.
    """
    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, *, registry=None)

Build the Recipe output lazily from named runtime inputs.

Parameters:

Name	Type	Description	Default
inputs	Mapping[str, Any]	Exact mapping from each :class:RecipeInput name to a Wandas Frame or NumPy/Dask array of the declared kind.	required
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 with semantic lineage, metadata, and Dask laziness
Any	preserved by its public operations. An identity plan with no nodes returns
Any	its input Frame unchanged.

Raises:

Type	Description
RecipeExecutionError	If names or runtime value kinds do not match, 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],
    *,
    registry: RecipeRegistry | None = None,
) -> Any:
    """Build the Recipe output 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.
        registry: Registry used for validation and operation execution. Supply
            the same extension registry used during extraction and loading.

    Returns:
        The output Frame 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, an
            operation fails, or an operation violates the Frame lineage contract.
        RecipeValidationError: If the plan is invalid for ``registry``.
    """
    return RecipeExecutor(registry=registry).execute(self, inputs)

Errors

wandas.pipeline.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."""

wandas.pipeline.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."""

wandas.pipeline.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."""

wandas.pipeline.RecipeExecutionError

Bases: RuntimeError

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

Source code in wandas/pipeline/errors.py

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

Extension declarations

wandas.pipeline.recipe_operation(operation_id, *, version=1, bindings=None, binding_patterns=None, capture=None, handler=None, validate_params=None)

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

Returns:

Type	Description
Callable[[Callable[P, R]], Callable[P, R]]	A method decorator that preserves the wrapped signature and exposes its
Callable[[Callable[P, R]], Callable[P, R]]	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,
) -> 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.

    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),
        )

        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

wandas.pipeline.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

wandas.pipeline.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)

wandas.pipeline.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

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.
    """

    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)

    def __post_init__(self) -> None:
        """Snapshot and validate the complete operation contract."""
        object.__setattr__(self, "binding_patterns", tuple(self.binding_patterns))
        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")

    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)

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

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 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")

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)

__init__(operation_id, version, binding_patterns, handler, validate_params=_no_param_validation)

wandas.pipeline.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

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

wandas.pipeline.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
RecipeRegistry	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())

Scikit-learn adapters

These stateless estimators call declared public Frame methods and preserve sklearn's clone and pipeline conventions.

wandas.pipeline.sklearn.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.

        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__}")
            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.

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.

    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__}")
        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

wandas.pipeline.sklearn.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")

wandas.pipeline.sklearn.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")

wandas.pipeline.sklearn.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")

wandas.pipeline.sklearn.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")

wandas.pipeline.sklearn.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")