Work with RecipePlan¶

For a guided, executable introduction, start with the Reusable Pipeline Recipes Learning Path. Use this how-to when you already know the task you need to complete.

RecipePlanを初めて段階的に学ぶ場合は、実行可能な Reusable Pipeline Recipes Learning Path から始めてください。具体的な作業が決まっている場合に、このHow-toを使います。

Choose WDF or Recipe¶

WDF saves one concrete, typed Frame result for later inspection or loading. Recipe saves reusable workflow intent so the same public operations can run on another input. A Recipe does not contain the original sample data; choose WDF when the result is the artifact and Recipe when the process is the artifact.

Saving or loading WDF needs the io extra; a Recipe file does not.

pip install "wandas[io]"

WDFの保存と読み込みにはio extraが必要です。Recipeファイルの保存には要りません。

What a Recipe stores¶

previous is a process-local reference to the immediately preceding Frame; it is not portable Recipe input.
lineage is the complete provenance of Frame inputs.
Runtime-only callables, regex predicates, and opaque Python objects cannot be serialized into a Recipe; extraction fails explicitly at that operation.

Named Frame and external-array inputs are supplied again at replay. The serialized artifact stores operation intent and bindings, not the input samples.

Set up the examples¶

The snippets below form one small workflow. Prepare a source, a replacement Frame, and an external array before choosing the task you need:

import numpy as np
import wandas as wd

source = wd.from_numpy(
    np.array([[1.0, 2.0, 4.0, 7.0]]),
    sampling_rate=8_000,
    ch_labels=["sensor"],
)
another_frame = wd.from_numpy(
    np.array([[2.0, 5.0, 8.0, 14.0]]),
    sampling_rate=8_000,
    ch_labels=["sensor"],
)
external_array = np.ones((1, 4))
processed = source.remove_dc().normalize()

Extract and apply a plan¶

Process a Frame normally, then name the runtime inputs while compiling its semantic lineage:

from wandas.pipeline import RecipePlan

processed = source.remove_dc().normalize()
plan = RecipePlan.from_frame(processed, input_names=("signal",))
replayed = plan.apply({"signal": another_frame})

The names supplied during extraction must match the mapping passed to apply(). A plan with one input also takes the Frame directly, which is the same as the mapping above:

replayed = plan.apply(another_frame)

A plan with several inputs still needs the mapping; passing one Frame raises RecipeExecutionError listing the input names. The result is a Frame; materialize it only at the boundary required by the task.

入力が1つのPlanは、plan.apply(another_frame)のようにFrameをそのまま渡せます。入力名の辞書で渡すのと同じです。入力が複数のPlanにFrameを1つだけ渡すと、入力名を示した RecipeExecutionErrorになります。

Save and load a standalone artifact¶

Save reusable operation intent separately from Frame data:

path = plan.save("analysis")  # analysis.recipe.json
restored = RecipePlan.load(path)
replayed = restored.apply(another_frame)

Use to_dict() and from_dict() when the artifact must travel through another JSON or storage layer. Saving does not overwrite an existing artifact unless that is requested explicitly. The format-specific schema and API errors are in the Pipeline API Reference.

Apply a saved Recipe to many recordings¶

apply_files reads each recording, applies a one-input plan, and saves each result as WDF (it needs pip install "wandas[io]"). A recording that fails is recorded and the rest go on; the returned pandas table has one row per recording with path, status, output, error, and seconds.

from pathlib import Path

plan = RecipePlan.load("analysis.recipe.json")
recordings = sorted(Path("recordings").rglob("*.wav"))
table = plan.apply_files(recordings, "outputs", root="recordings", jobs=4)
table.to_csv("outputs/results.csv", index=False)
print(table[table["status"] == "error"])

root keeps each recording's folder below it (recordings/L1/a.wav becomes outputs/L1/a.wav.wdf). jobs runs recordings in separate processes; a script that uses it needs an if __name__ == "__main__": guard. Pass on_error="raise" to stop at the first failure (with jobs, later recordings may already be saved), and overwrite=True to replace earlier outputs.

保存したRecipeを多数の録音に掛けるにはapply_filesを使います。録音ごとに読み込み・実行・WDF保存を行い、失敗した録音は記録して先に進みます。戻り値は録音ごとの成否・出力パス・エラー・所要秒の表（pandas）です。rootを渡すとその下のフォルダ構成を保ち、jobsで別プロセスに分けて並列に実行します（スクリプトではif __name__ == "__main__":が必要です）。on_error="raise"は最初の失敗で止めます（jobsを使うと、後の録音がすでに保存されていることがあります）。overwrite=Trueで既存の出力を置き換えます。

Name steps and choose which one to return¶

A plan returns one Frame per apply. To keep shared preprocessing and several analyses of it in one Recipe, name the other Frames as steps when you extract the plan, then choose one by name:

signal = wd.generate_sin(freqs=[1000.0], duration=0.5)
another_signal = wd.generate_sin(freqs=[500.0], duration=0.5)
weighted = signal.high_pass_filter(cutoff=50).a_weighting()
plan = RecipePlan.from_frame(
    weighted.welch(),
    steps={"weighted": weighted, "noct": weighted.noct_spectrum(fmin=100, fmax=5_000)},
)

spectrum = plan.apply(another_signal)              # the output (welch)
octaves = plan.apply(another_signal, step="noct")  # only noct is built
plan.step_names                                    # ('weighted', 'noct')

The shared filter and A-weighting are stored once. apply(..., step=name) builds only the operations that step needs, so steps you do not ask for are neither computed nor kept in memory. Each call computes its own preprocessing: asking for welch and then noct filters the input twice. Steps must come from the same source Frames as the output.

apply_files(..., step="noct") saves that step for each recording, and plan.for_step("noct") returns a plan with only that step's operations and inputs to save or share on its own. apply(..., step=...) still takes every input of the plan. A plan with steps is saved as Recipe schema 4; a plan without steps is still saved as schema 3.

Planがapplyごとに返すFrameは1つです。共通の前処理とそれを使う複数の解析を1つのRecipeにまとめるには、抽出するときに他のFrameをstepsとして名前付きで渡し、apply(..., step=名前)で返す段を選びます。共通の前処理は1回だけ記録されます。選んだ段に必要な処理だけを組み立てるので、選ばない段は計算もメモリへの保持もされません。ただし呼び出しごとに前処理を計算するので、 welchとnoctを続けて求めると前処理は2回計算されます。段は出力と同じ元のFrameから作ってください。 apply_files(..., step="noct")は録音ごとにその段を保存し、plan.for_step("noct")はその段の処理と入力だけを持つPlanを返します（apply(..., step=...)にはPlanのすべての入力を渡します）。段を持つPlanはRecipe schema 4で保存され、段のないPlanはこれまでどおりschema 3です。

Supply multiple inputs¶

Give every runtime Frame or external array a stable name. Input order and alignment are defined by the public operation that created the plan.

processed = source + external_array
plan = RecipePlan.from_frame(processed, input_names=("signal", "offset"))
replayed = plan.apply({"signal": another_frame, "offset": external_array})

External NumPy and Dask arrays remain named inputs; they are not embedded in the artifact or wrapped in temporary Frames. At replay, supply a NumPy or Dask array with a shape and dtype accepted by the operation. The original container type and Dask chunking are not stored in the Recipe.

List operations and their parameters¶

RecipeRegistry.describe() returns one description per registered operation: its ID, version, method name, input bindings, and the method parameters with their defaults. Parameters annotated with Literal[...] also report their accepted values as choices. Use it to build parameter forms, documentation, or tool schemas from the same registry that validates and replays Recipes.

from wandas.pipeline import default_recipe_registry

catalog = default_recipe_registry().describe()
fft = next(item for item in catalog if item.method == "fft")
window = next(parameter for parameter in fft.parameters if parameter.name == "window")
print(window.default)  # hann

Contracts kept only to replay older Recipes are listed too. 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__.

Each description also says which Frame types the operation is called on (receiver), what it returns (returns; "receiver" when the result keeps the caller's type, such as filters), and which optional extras it needs (extras, as in pip install "wandas[psychoacoustic]"). applies_to(frame) checks one Frame, including a subclass that overrides the method with another operation. For example, list what a SpectralFrame accepts, leaving out operations whose extras are missing:

from importlib.util import find_spec

needs = {"effects": "librosa", "psychoacoustic": "mosqito"}
spectrum = source.fft()
usable = [
    item.method
    for item in catalog
    # Skip replay-only contracts (one leading "_"); keep operators such as __add__.
    if not (item.method.startswith("_") and not item.method.startswith("__"))
    and item.applies_to(spectrum)
    and all(find_spec(needs[extra]) for extra in item.extras)
]

RecipeRegistry.describe()は登録済み操作ごとに、ID、version、メソッド名、入力の binding、メソッドの引数と既定値を返します。Literal[...]で注釈した引数は受け付ける値を choicesとして返します。パラメータ入力欄、ドキュメント、ツール定義を、Recipeの検証と再実行に使うのと同じregistryから作れます。古いRecipeの再実行用に残した操作も含まれ、そのmethodは _fft_recipe_v2のように先頭が_1つの非公開名です。公開呼出しが記録するのはmethodが公開名の操作で、__add__のような演算子メソッドも含みます。各説明には、その操作を呼べるFrameの型（receiver）、結果の型（returns。フィルタのように呼んだFrameと同じ型なら"receiver"）、必要なextra（extras）も入ります。applies_to(frame)は、メソッドを別の操作で上書きしたサブクラスも含めて、あるFrameにその操作を呼べるかを判定します。あるSpectralFrameに何を呼べるかを調べたり、extraが入らない環境で使える操作を前もって絞ったりできます。

Annotate each runtime input¶

Labels, metadata and channel extra describe one recording, so a plan does not record with_label(), with_metadata() or with_channel_extra(); replay uses the annotations of the frame you pass in. A plan that selects channels by extra, such as get_channel(query={"position": "front"}), therefore needs the same annotation on every runtime input, for example from that recording's catalog row. This example uses two-channel recordings of its own:

recording = wd.from_numpy(
    np.random.default_rng(0).normal(size=(2, 800)),
    sampling_rate=8_000,
    ch_labels=["mic1", "mic2"],
)
annotated = recording.with_channel_extra("mic2", {"position": "front"})
front = annotated.get_channel(query={"position": "front"}).high_pass_filter(cutoff=100)
plan = RecipePlan.from_frame(front, input_names=("signal",))

next_recording = wd.from_numpy(
    np.random.default_rng(1).normal(size=(2, 800)),
    sampling_rate=8_000,
    ch_labels=["mic1", "mic2"],
)
other = next_recording.with_channel_extra("mic2", {"position": "front"})
replayed = plan.apply({"signal": other})  # selects mic2

Replaying on a frame where no channel has the position key raises RecipeExecutionError naming the unknown metadata key. If the key exists but no channel has the value "front", the error says that no channel matches the query. If a different channel carries "front", replay selects that channel, so annotate each input from its own catalog row.

ラベル・metadata・チャネルのextraは録音ごとの情報なので、Recipeには記録されず、再生時は渡したフレームの注釈が使われます。get_channel(query={"position": "front"})のように extraでチャネルを選ぶRecipeは、再生する各入力にも同じ注釈（たとえばその録音の諸元表の値）を付けてから渡してください。どのチャネルにもそのキーがない入力では、未知のメタデータキーを示す RecipeExecutionErrorになります。キーはあっても値が一致するチャネルがなければ、クエリに一致するチャネルがないというエラーになります。別のチャネルに同じ値を付けると、そのチャネルが選ばれるので、注釈は各録音の諸元表から付けてください。

Find the step that failed¶

When an operation fails during apply, the RecipeExecutionError carries the failing node as attributes, so a tool can mark that step without parsing the message. node_id and operation are None for errors that do not belong to one node, such as mismatched input names.

from wandas.pipeline import RecipeExecutionError

try:
    plan.apply({"signal": source})
except RecipeExecutionError as error:
    print(error.node_id, error.operation, repr(error.cause))

applyで処理が失敗すると、RecipeExecutionErrorのnode_id・operation・cause （元の例外、__cause__と同じ）で失敗したノードが分かります。エラー文を読まずに、失敗した手順へ印を付けられます。入力名の不一致のようにノードに関係しないエラーでは、 node_idとoperationはNoneです。

Handle runtime-only operations¶

Some calls intentionally cannot be portable. Arbitrary callables passed to Frame.apply(), callable or regex channel predicates, and opaque Python objects must remain runtime-only. Recipe extraction fails at the unsupported operation instead of silently dropping part of the workflow.

To keep a portable operation in your own project, follow Add your own processing, including the registry passed to extraction, loading, and replay. 自分のプロジェクト内でRecipe対応の独自処理を作る場合は、独自処理のHow-toを参照してください。

For built-in contributions and advanced Frame contracts, use the Frame and Operation extension guide.