Pipeline Operator - Functional Composition#

The pipeline operator (|) is one of TorchFX’s most distinctive features. It provides an intuitive, readable way to chain audio effects together, creating complex processing pipelines with simple, linear code.

What is the Pipeline Operator?#

The pipeline operator is Python’s bitwise OR operator (|) repurposed for functional composition. In TorchFX, it means “apply this effect to the audio”:

import torchfx as fx

wave = fx.Wave.from_file("audio.wav")

# Read this as: "wave, apply lowpass filter"
filtered = wave | fx.filter.iir.LoButterworth(1000)

This approach is inspired by Unix pipes and functional programming languages, making audio processing code read like a natural description of the signal flow.

        graph LR
    A[Input Wave] -->|"|"| B[Effect 1]
    B -->|"|"| C[Effect 2]
    C -->|"|"| D[Effect 3]
    D --> E[Output Wave]

    style A fill:#e1f5ff
    style E fill:#e1f5ff
    style B fill:#fff5e1
    style C fill:#fff5e1
    style D fill:#fff5e1

Basic Usage#

Single Effect#

Apply one effect to a wave:

import torchfx as fx
from torchfx.filter import iir

wave = fx.Wave.from_file("audio.wav")

# Apply a single effect
filtered = wave | iir.LoButterworth(cutoff=1000, order=4)

Chaining Effects#

Chain multiple effects sequentially:

import torchfx as fx
from torchfx.filter import iir

wave = fx.Wave.from_file("audio.wav")

# Chain multiple effects: read left to right
processed = wave | iir.HiButterworth(100) | iir.LoButterworth(5000) | fx.effect.Normalize()

This is equivalent to:

step1 = wave | iir.HiButterworth(100)
step2 = step1 | iir.LoButterworth(5000)
processed = step2 | fx.effect.Normalize()

But much more concise and readable.

Multi-line Pipelines#

For complex pipelines, use parentheses for multi-line formatting:

import torchfx as fx
from torchfx.filter import iir

wave = fx.Wave.from_file("audio.wav")

processed = (
    wave
    | iir.HiButterworth(cutoff=80, order=2)      # Remove subsonic rumble
    | iir.LoButterworth(cutoff=12000, order=4)   # Remove high-frequency noise
    | fx.effect.Normalize(peak=0.9)               # Normalize to -0.9 dB
    | fx.effect.Reverb(delay=4410, decay=0.3)    # Add subtle reverb
)

processed.save("output.wav")

How It Works#

The `or` Method#

The pipeline operator is implemented via Python’s __or__ magic method in the Wave class:

# Simplified implementation from torchfx/wave.py
class Wave:
    def __or__(self, f: nn.Module) -> "Wave":
        """Apply a module to the wave through the pipeline operator."""
        # Validate input
        if not isinstance(f, nn.Module):
            raise TypeError(f"Expected nn.Module, but got {type(f).__name__}")

        # Auto-configure FX effects
        if isinstance(f, FX):
            self.__update_config(f)

        # Apply effect and return new Wave
        return self.transform(f.forward)

Automatic Configuration#

When you use the pipeline operator, TorchFX automatically:

Validates the effect: Ensures it’s a torch.nn.Module
Sets the sample rate: Configures the effect’s fs attribute from the wave
Computes coefficients: For filters, calls compute_coefficients() if needed
Applies the effect: Calls the effect’s forward() method on the audio tensor
Returns a new Wave: Preserves immutability

        sequenceDiagram
    participant User
    participant Wave
    participant Effect
    participant Tensor

    User->>Wave: wave | effect
    Wave->>Wave: __or__(effect)
    Wave->>Effect: Check if FX instance
    Wave->>Effect: Set effect.fs = wave.fs
    Wave->>Effect: Call compute_coefficients() (if filter)
    Wave->>Effect: Call effect.forward(wave.ys)
    Effect->>Tensor: Process audio tensor
    Tensor-->>Effect: Processed tensor
    Effect-->>Wave: Return processed tensor
    Wave->>Wave: Create new Wave(processed, fs)
    Wave-->>User: Return new Wave

Immutability#

The pipeline operator returns a new Wave object, leaving the original unchanged:

import torchfx as fx

original = fx.Wave.from_file("audio.wav")
filtered = original | fx.filter.iir.LoButterworth(1000)

# Original is unchanged
assert len(original) == len(original.ys)

# Filtered is a new object
assert filtered is not original

This functional programming pattern:

Prevents accidental mutations
Makes debugging easier
Enables safe parallel processing
Supports method chaining

Advanced Patterns#

Combining with PyTorch Modules#

The pipeline operator works with any torch.nn.Module, not just FX effects:

import torch
import torchfx as fx

wave = fx.Wave.from_file("audio.wav")

# Use PyTorch Sequential
effect_chain = torch.nn.Sequential(
    fx.filter.iir.HiButterworth(100),
    fx.filter.iir.LoButterworth(5000),
    fx.effect.Normalize()
)

processed = wave | effect_chain

Parallel Filter Combination#

Use the + operator to combine filters in parallel:

import torchfx as fx
from torchfx.filter import iir

wave = fx.Wave.from_file("audio.wav")

# Create bandpass filter: parallel high-pass + low-pass
bandpass = iir.HiButterworth(200) + iir.LoButterworth(2000)

# Apply combined filter
filtered = wave | bandpass

The + operator creates a ParallelFilterCombination that:

Applies both filters to the input
Sums the outputs
Returns the combined result

        graph TB
    Input[Input Signal] --> Split{Split}
    Split --> HP[High-Pass<br/>200 Hz]
    Split --> LP[Low-Pass<br/>2000 Hz]
    HP --> Sum((+))
    LP --> Sum
    Sum --> Output[Output Signal]

    style Input fill:#e1f5ff
    style Output fill:#e1f5ff
    style HP fill:#fff5e1
    style LP fill:#fff5e1

Reusable Effect Chains#

Define effect chains once and reuse them:

import torchfx as fx
from torchfx.filter import iir
import torch.nn as nn

# Define a reusable mastering chain
mastering_chain = nn.Sequential(
    iir.HiButterworth(cutoff=30, order=2),     # Remove DC offset and rumble
    iir.LoButterworth(cutoff=18000, order=4),  # Remove ultrasonic content
    fx.effect.Normalize(peak=0.95),            # Normalize
)

# Apply to multiple files
for audio_file in ["track1.wav", "track2.wav", "track3.wav"]:
    wave = fx.Wave.from_file(audio_file)
    mastered = wave | mastering_chain
    mastered.save(f"mastered_{audio_file}")

Conditional Processing#

Apply effects conditionally:

import torchfx as fx

def process_audio(wave: fx.Wave, apply_reverb: bool = False) -> fx.Wave:
    # Always apply EQ
    processed = wave | fx.filter.iir.LoButterworth(5000)

    # Conditionally apply reverb
    if apply_reverb:
        processed = processed | fx.effect.Reverb(delay=4410, decay=0.4)

    # Always normalize
    processed = processed | fx.effect.Normalize()

    return processed

wave = fx.Wave.from_file("audio.wav")
dry = process_audio(wave, apply_reverb=False)
wet = process_audio(wave, apply_reverb=True)

Per-Channel Processing#

Process individual channels separately:

import torchfx as fx

wave = fx.Wave.from_file("stereo.wav")  # 2 channels

# Extract and process left channel
left = wave.get_channel(0) | fx.effect.Delay(delay_samples=4410)

# Extract and process right channel differently
right = wave.get_channel(1) | fx.effect.Delay(delay_samples=2205)

# Merge back to stereo
stereo = fx.Wave.merge([left, right], split_channels=True)

Comparison with Other Approaches#

Pipeline Operator (TorchFX)#

import torchfx as fx

wave = fx.Wave.from_file("audio.wav")
processed = wave | Effect1() | Effect2() | Effect3()

Advantages:

Reads left to right (natural flow)
Concise and expressive
Auto-configures effects
Type-safe (IDE support)

Sequential Container (PyTorch)#

import torch.nn as nn

effects = nn.Sequential(Effect1(), Effect2(), Effect3())
processed_tensor = effects(audio_tensor)

Advantages:

Standard PyTorch pattern
Works with neural networks
Explicit container

Disadvantages:

Need to manage sample rate manually
Returns tensor, not Wave
More verbose

Manual Chaining#

x = audio_tensor
x = effect1(x)
x = effect2(x)
x = effect3(x)
processed_tensor = x

Advantages:

Explicit and clear
Easy to debug intermediate steps

Disadvantages:

Verbose
Need to manage variables
Easy to make mistakes

Best Practices#

Use Multi-line for Readability#

For complex pipelines, format across multiple lines:

# ✅ GOOD: Multi-line with comments
processed = (
    wave
    | iir.HiButterworth(80, order=2)     # Remove rumble
    | iir.LoButterworth(12000, order=4)  # Remove noise
    | fx.effect.Normalize(peak=0.9)      # Normalize
)

# ❌ BAD: Everything on one line
processed = wave | iir.HiButterworth(80, order=2) | iir.LoButterworth(12000, order=4) | fx.effect.Normalize(peak=0.9)

Extract Complex Chains#

For reusable or complex chains, use torch.nn.Sequential:

import torch.nn as nn
import torchfx as fx

# ✅ GOOD: Named, reusable chain
vocal_chain = nn.Sequential(
    iir.HiButterworth(cutoff=100, order=2),
    iir.PeakingEQ(freq=3000, gain_db=3, q=1.0),  # Presence boost
    fx.effect.Compressor(threshold=0.5, ratio=4.0),
    fx.effect.Normalize(peak=0.95),
)

processed = wave | vocal_chain

# ❌ LESS GOOD: Inline complex chain
processed = wave | iir.HiButterworth(100, 2) | iir.PeakingEQ(3000, 3, 1.0) | fx.effect.Compressor(0.5, 4.0) | fx.effect.Normalize(0.95)

Avoid Deep Nesting#

Keep pipelines flat for readability:

# ✅ GOOD: Flat pipeline
processed = wave | Effect1() | Effect2() | Effect3()

# ❌ BAD: Nested pipelines (confusing)
processed = wave | (Effect1() | (Effect2() | Effect3()))

Name Intermediate Results#

For debugging or visualization, name intermediate steps:

import torchfx as fx

wave = fx.Wave.from_file("audio.wav")

# Name each step for debugging
highpassed = wave | fx.filter.iir.HiButterworth(100)
lowpassed = highpassed | fx.filter.iir.LoButterworth(5000)
normalized = lowpassed | fx.effect.Normalize()

# Can save intermediate results
highpassed.save("debug_highpass.wav")
lowpassed.save("debug_lowpass.wav")
normalized.save("output.wav")

Common Pitfalls#

Modifying Effects After Application#

Effects are configured when applied to the wave. Modifying them afterward has no effect:

import torchfx as fx

wave = fx.Wave.from_file("audio.wav")
effect = fx.effect.Gain(gain=1.0)

processed = wave | effect

# ❌ This doesn't change the processed wave
effect.gain = 2.0

# ✅ Create a new effect and apply it
processed = wave | fx.effect.Gain(gain=2.0)

Forgetting to Assign Result#

The pipeline operator returns a new Wave; forgetting to assign loses the result:

import torchfx as fx

wave = fx.Wave.from_file("audio.wav")

# ❌ WRONG: Result is lost
wave | fx.effect.Normalize()

# ✅ CORRECT: Assign result
normalized = wave | fx.effect.Normalize()

# ✅ ALSO CORRECT: Reassign to same variable (if you don't need original)
wave = wave | fx.effect.Normalize()

Mixing Sample Rates#

All effects in a chain should work with the same sample rate:

import torchfx as fx

wave_44k = fx.Wave.from_file("audio_44100.wav")  # 44100 Hz
wave_48k = fx.Wave.from_file("audio_48000.wav")  # 48000 Hz

# ❌ WRONG: Effects configured for different rates
effect_44k = wave_44k | fx.effect.Delay(bpm=120, delay_time="1/4")
processed_48k = wave_48k | effect_44k  # Effect still has fs=44100!

# ✅ CORRECT: Each wave configures its own effects
processed_44k = wave_44k | fx.effect.Delay(bpm=120, delay_time="1/4")
processed_48k = wave_48k | fx.effect.Delay(bpm=120, delay_time="1/4")

Performance Considerations#

In-Place vs. Immutable#

The pipeline operator creates new Wave objects. For memory-constrained environments, consider:

import torchfx as fx

wave = fx.Wave.from_file("audio.wav")

# Memory-friendly: reuse variable name
wave = wave | Effect1()
wave = wave | Effect2()
wave = wave | Effect3()
# Old Wave objects are garbage collected

# Memory-intensive: keep all intermediates
step1 = wave | Effect1()
step2 = step1 | Effect2()
step3 = step2 | Effect3()
# All Wave objects remain in memory

GPU Acceleration#

Move waves to GPU before applying effects:

import torchfx as fx
import torch

if torch.cuda.is_available():
    wave = fx.Wave.from_file("audio.wav").to("cuda")

    # Effects run on GPU
    processed = wave | HeavyEffect1() | HeavyEffect2()

    # Move back to CPU for saving
    processed.to("cpu").save("output.wav")
else:
    wave = fx.Wave.from_file("audio.wav")
    processed = wave | HeavyEffect1() | HeavyEffect2()
    processed.save("output.wav")

Batch Processing#

Reuse effect instances for batch processing:

import torchfx as fx
from pathlib import Path

# Create effects once
effect_chain = fx.effect.Normalize() | fx.effect.Reverb()

# Reuse for multiple files
for audio_file in Path("audio").glob("*.wav"):
    wave = fx.Wave.from_file(audio_file)
    processed = wave | effect_chain
    processed.save(f"processed/{audio_file.name}")

Mathematical Interpretation#

The pipeline operator can be understood as function composition. Given effects \(f\), \(g\), and \(h\):

result = wave | f | g | h

Is equivalent to:

\[ \text{result} = h(g(f(\text{wave}))) \]

Or in mathematical notation with the composition operator \(\circ\):

\[ \text{result} = (h \circ g \circ f)(\text{wave}) \]

But TorchFX’s syntax reads left-to-right (natural order) instead of right-to-left (mathematical order):

\[ \text{wave} \xrightarrow{f} \text{temp}_1 \xrightarrow{g} \text{temp}_2 \xrightarrow{h} \text{result} \]

External Resources#

Unix Pipes on Wikipedia - Inspiration for the pipeline pattern
Fluent Interface on Wikipedia - Design pattern for method chaining
Python Magic Methods - Understanding __or__

Pipeline Operator - Functional Composition#

What is the Pipeline Operator?#

Basic Usage#

Single Effect#

Chaining Effects#

Multi-line Pipelines#

How It Works#

The `or` Method#

Automatic Configuration#

Immutability#

Advanced Patterns#

Combining with PyTorch Modules#

Parallel Filter Combination#

Reusable Effect Chains#

Conditional Processing#

Per-Channel Processing#

Comparison with Other Approaches#

Pipeline Operator (TorchFX)#

Sequential Container (PyTorch)#

Manual Chaining#

Best Practices#

Use Multi-line for Readability#

Extract Complex Chains#

Avoid Deep Nesting#

Name Intermediate Results#

Common Pitfalls#

Modifying Effects After Application#

Forgetting to Assign Result#

Mixing Sample Rates#

Performance Considerations#

In-Place vs. Immutable#

GPU Acceleration#

Batch Processing#

Mathematical Interpretation#

External Resources#

References#

This Page

Pipeline Operator - Functional Composition#

What is the Pipeline Operator?#

Basic Usage#

Single Effect#

Chaining Effects#

Multi-line Pipelines#

How It Works#

The __or__ Method#

Automatic Configuration#

Immutability#

Advanced Patterns#

Combining with PyTorch Modules#

Parallel Filter Combination#

Reusable Effect Chains#

Conditional Processing#

Per-Channel Processing#

Comparison with Other Approaches#

Pipeline Operator (TorchFX)#

Sequential Container (PyTorch)#

Manual Chaining#

Best Practices#

Use Multi-line for Readability#

Extract Complex Chains#

Avoid Deep Nesting#

Name Intermediate Results#

Common Pitfalls#

Modifying Effects After Application#

Forgetting to Assign Result#

Mixing Sample Rates#

Performance Considerations#

In-Place vs. Immutable#

GPU Acceleration#

Batch Processing#

Mathematical Interpretation#

Related Concepts#

External Resources#

References#

This Page

The `or` Method#