pyFDN.td package#

Submodules#

pyFDN.td.connectors module#

class pyFDN.td.connectors.Parallel(ops, *, sum_output=True)[source]#

Bases: TimeOperator

Feed the same input to every branch and combine the outputs (optional). Equivalent to FLAMO Parallel.

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

reset()[source]#

Clear internal state (no-op for stateless operators).

Return type:: None

class pyFDN.td.connectors.Recursion(forward, feedback, *, block_size, delay_position='forward')[source]#

Bases: TimeOperator

Closed feedback loop, y[n] = fF(x[n] + fB(y[n - block_size])).

A feedback loop cannot be evaluated sample by sample without an algebraic loop, so Recursion computes whole blocks of block_size samples at a time: it reads the loop state written by the previous block, runs both paths over the block, and writes the result back. That read-before-write is what breaks the loop – and it inserts exactly ``block_size`` samples of delay into the loop, on top of whatever delay the operators themselves have.

The inserted delay is not compensated automatically; a warning is issued at construction as a reminder. To compensate, shorten the delay lines on the path named by delay_position by block_size samples, e.g. Delay(delays - block_size) – which requires every delay on that path to be at least 2 * block_size long, since the shortened line still has to be at least one block long.

Parameters:

forward (TimeOperator) – Forward path fF, from the loop input to the loop output.
feedback (TimeOperator) – Feedback path fB, from the loop output back to the loop input.
block_size (int) – Processing block size in samples, and therefore the amount of delay inserted into the loop. Required: it is a property of the loop the caller has to choose and compensate for, not an implementation detail. It must not exceed the shortest delay on the loop, or the loop would run ahead of its own delay lines.
delay_position (str) –
Which side of the loop the inserted block_size delay lands on, i.e. where the state buffer sits:

"forward" (default)
After the forward path: the loop output is read out of the state buffer, so it is the forward output that arrives block_size samples late (y[n] = fF(...)[n - block_size]). Compensate on the forward path. Use this when the delay lines are in the forward path (the usual FDN layout: delays forward, mixing matrix feedback).

"feedback"
After the feedback path: the forward output is returned immediately and it is the signal fed back that is block_size samples late (y[n] = fF(x[n] + fB(y)[n - block_size])). Compensate on the feedback path. Use this when the delay lines are in the feedback path, e.g. an outer acoustic-feedback loop around a whole system.

Both give the same total loop delay; they differ in where in the loop that delay sits, hence in which path has to absorb the compensation and whether the operator’s own output is delayed.

filter(block)[source]#

Filter block of audio.

Parameters:: block (ArrayLike) – Input block of audio.
Raises:: ValueError – Mismatch between audio-block channels and Recursion input channels
Returns:: Output block of audio.
Return type:: ndarray

reset()[source]#

Resets the state of all internal TimeOperators.

Return type:: None

class pyFDN.td.connectors.Series(ops)[source]#

Bases: TimeOperator

Chain operators left to right. Equivalent to FLAMO Series.

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

reset()[source]#

Clear internal state (no-op for stateless operators).

Return type:: None

pyFDN.td.operators module#

Stateful time-domain operators.

Each operator maps a block (num_samples, in_channels) to (num_samples, out_channels) and keeps whatever state it needs across calls, so a long signal can be streamed through in consecutive blocks. The composites (Series, Parallel, Recursion) live in pyFDN.td.connectors and wire these leaves into a graph.

RecursionState is not an operator but the delay-line buffer bank the graph is built on: it is what Recursion uses to break its feedback loop, and what pyFDN.process_fdn() uses for the FDN delay lines.

class pyFDN.td.operators.AbsoluteValue(channels)[source]#

Bases: TimeOperator

Stateless memoryless nonlinearity y[n, c] = |x[n, c]|.

The one non-linear operator in the set: it has no transfer function and no frequency-domain counterpart, so a graph containing it can only be rendered in the time domain. Useful as a rectifier inside a loop (distortion, envelope-style feedback) and as the simplest way to test that the block engine stays sample-exact when superposition no longer holds – note that a non-linear operator makes the render depend on the input level.

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

class pyFDN.td.operators.ControllableFullWaveRect(channels, alpha, active_channels)[source]#

Bases: TimeOperator

Stateful, controllable, memoryless nonlinearity, y[n, c] = g_cfwr * ((1 - alpha) * x[n, c] + alpha * abs(x[n, c])), applied to active_channels only; the rest pass through unchanged.

At alpha = 0 the nonlinearity drops out and only the DC blocker below is left, at alpha = 1 it is a full-wave rectifier; g_cfwr = sqrt(2 - 2 * abs(alpha - 0.5)) keeps the output power roughly constant across alpha. abs(x) here is not the plain absolute value but its first-order antiderivative-antialiasing approximation (Parker et al. 2016), which reduces the aliasing that rectification would otherwise fold back from above Nyquist. The result is passed through an internal DCBlocker with energy compensation, since rectification also injects a DC offset that would otherwise accumulate in a feedback loop.

abs(x)[source]#

Return type:: ndarray

anti_dev(x)[source]#

Return type:: ndarray

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

reset()[source]#

Clear internal state (no-op for stateless operators).

Return type:: None

class pyFDN.td.operators.DCBlocker(channels, R=0.995, correct_loss=False, fs=48000.0, env_tau_s=0.05, gain_tau_s=0.02, max_gain=4.0)[source]#

Bases: TimeOperator

Stateful per-channel first-order DC blocker with optional slow energy compensation, y[n, c] = x[n, c] - x[n-1, c] + R * y[n-1, c].

The differencing term removes the DC offset a nonlinearity such as ControllableFullWaveRect would otherwise inject into a feedback loop, at the cost of also attenuating content near DC. When correct_loss is enabled, a slowly-varying gain tracks the ratio of input to output power through two exponential envelope followers – one over the signal power, one smoothing the resulting gain – and rescales the output to compensate for that loss.

Parameters:

channels (int) – Number of channels processed independently.
R (float) – Pole location of the blocker, 0 < R < 1. Closer to 1 pushes the cutoff frequency down and preserves more low-frequency content.
correct_loss (bool) – If True, apply the energy-compensation gain described above.
fs (float) – Sampling rate in Hz, used to convert the time constants below into per-sample smoothing coefficients.
env_tau_s (float) – Time constant of the power envelope followers, in seconds.
gain_tau_s (float) – Time constant of the gain smoothing, in seconds.
max_gain (float) – Ceiling on the compensation gain. A signal the blocker removes almost entirely – anything close to pure DC – has an output power near zero, so the uncapped ratio grows without bound: it would undo the blocking it is compensating for and run away inside a feedback loop.

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

reset()[source]#

Clear internal state (no-op for stateless operators).

Return type:: None

class pyFDN.td.operators.Delay(delays)[source]#

Bases: TimeOperator

Stateful per-channel integer feed-forward delay line, y[n, c] = x[n - m_c, c]. It is not equivalent to RecursionState, as it does not accept direct feedback synchronization.

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

reset()[source]#

Clear internal state (no-op for stateless operators).

Return type:: None

class pyFDN.td.operators.Gain(matrix)[source]#

Bases: TimeOperator

Stateless static gain matrix y = x @ M.T with M of shape (out, in).

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

class pyFDN.td.operators.GranularPitchShift(channels, max_delay_samps, grain_dur_samps, transpose_cents, active_channels, fade_ratio=0.25, seed=None)[source]#

Bases: TimeOperator

Stateful, controllable granular pitch shifter.

Two grains, triggered half a grain apart, read a shared per-channel circular buffer at a rate set by transpose_cents. Each grain is windowed by a raised-cosine envelope and, on reaching grain_dur_samps, restarts at a new random position inside the buffer – trading the continuous read-head wraparound of PitchShift for grain-boundary clicks disguised by the envelope. Only active_channels are shifted; the rest pass through unchanged.

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

reset()[source]#

Clear internal state (no-op for stateless operators).

Return type:: None

set_transpose_cents(cents)[source]#

Return type:: None

class pyFDN.td.operators.Identity(channels)[source]#

Bases: TimeOperator

Stateless pass-through. Used as a placeholder branch, e.g. an empty forward residual.

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

class pyFDN.td.operators.MatrixConvolver(coeffs)[source]#

Bases: TimeOperator

Stateful matrix of FIR filters via streaming overlap-save FFT convolution. The FFT counterpart of MatrixFIR: same (n_out, n_in, n_taps) coefficient layout, but built for long impulse responses (e.g. room RIRs) where time-domain lfilter would be prohibitively slow. State persists across filter() calls, so it works both whole-signal and block-by-block – including as the feedback path of a Recursion (the loudspeaker -> microphone room coupling of a reverberation enhancement system). Output equals the linear convolution to numerical precision.

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

reset()[source]#

Clear internal state (no-op for stateless operators).

Return type:: None

class pyFDN.td.operators.MatrixFIR(coeffs)[source]#

Bases: TimeOperator

Stateful matrix of FIR filters (e.g. a paraunitary scattering feedback matrix).

Every matrix entry is an FIR filter run with scipy.signal.lfilter(); state persists across filter() calls. For long impulse responses use MatrixConvolver instead, which computes the same convolution by FFT.

Parameters:: coeffs (ArrayLike) – FIR coefficients of shape (n_out, n_in, n_taps) in the z^{-1} convention (coeffs[i, j, k] is the tap of z^{-k} from input j to output i).

Notes

pyFDN.process_fdn() constructs this filter automatically when its feedback matrix A has shape (n_out, n_in, n_taps).

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

reset()[source]#

Clear internal state (no-op for stateless operators).

Return type:: None

class pyFDN.td.operators.PitchShift(channels, max_delay_samps, window_size, transpose_cents, fs, active_channels, min_delay_samps=3)[source]#

Bases: TimeOperator

Stateful, controllable dual-read-head pitch shifter.

Writes into one circular buffer per channel and reads it back with two read heads spaced half a window apart, each sliding at a rate set by transpose_cents and crossfaded with a complementary sine window so the head that is about to wrap is always faded out. Only active_channels are shifted; the rest pass through unchanged.

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

reset()[source]#

Clear internal state (no-op for stateless operators).

Return type:: None

set_transpose_cents(cents)[source]#

Return type:: None

class pyFDN.td.operators.RecursionState(delays, max_block_size)[source]#

Bases: object

Vectorised bank of block-addressable delay lines.

The state store a feedback loop is built on: get_values() reads the next block_size output samples of every line, set_values() writes the samples going back in, and advance() moves the read/write pointers. Reading before writing is what breaks the algebraic loop, so a whole block can be computed at once.

Used by Recursion (all lines the length of one processing block) and by pyFDN.process_fdn() (one line per FDN delay).

Parameters:

delays (ArrayLike) – Length of each delay line in samples, shape (num_delays,). Must be positive.
max_block_size (int) – Largest block get_values() will be asked for. Must not exceed the shortest delay, otherwise a block would wrap around its own line.

advance(block_size)[source]#

Move the read/write pointers on by one block.

Return type:: None

get_values(block_size)[source]#

Read the next block_size samples out of every delay line.

Return type:: ndarray

reset()[source]#

Zero the buffers and rewind the pointers.

Return type:: None

set_values(block)[source]#

Write a (block_size, num_delays) block into the slots just read.

Return type:: None

class pyFDN.td.operators.RingModulator(channels, mod_freq, mod_amp, fs, active_channels)[source]#

Bases: TimeOperator

Stateful, controllable ring modulator, y[n, c] = mod_amp * x[n, c] * sin(2 * pi * mod_freq * n / fs), applied to active_channels only; the rest pass through unchanged.

Multiplying by a sine shifts the spectrum of the active channels by +-mod_freq rather than adding harmonics on top of it, so the effect reads as tremolo at low mod_freq and as an inharmonic, bell-like retuning once mod_freq enters the audible range. A unit-amplitude sine carries only half the power of the signal it multiplies, so mod_amp = sqrt(2) keeps the operator energy-preserving. The modulation phase runs continuously across filter() calls, tracked by sample_index.

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

reset()[source]#

Clear internal state (no-op for stateless operators).

Return type:: None

class pyFDN.td.operators.SDFD(channels, d, active_channels)[source]#

Bases: TimeOperator

Stateful, controllable Signal-Dependent Fractional Delay, y[n, c] = (1 - d) n[n-1, c] + d n[n, c] + d p[n-2, c] + (1 - d) p[n-1, c], applied to active_channels only; the rest pass through unchanged.

p = max(x, 0) and n = min(x, 0) are the positive and negative half-wave rectified branches of the input, each delayed by a different, d-dependent amount: the positive branch by roughly 1 + d samples, the negative one by roughly 1 - d. Recombining the two smears the signal’s zero crossings without reshaping the rest of the waveform, which reads as a soft, amplitude-dependent distortion rather than a hard clip.

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

reset()[source]#

Clear internal state (no-op for stateless operators).

Return type:: None

class pyFDN.td.operators.SOSBank(sos)[source]#

Bases: TimeOperator

Stateful per-channel SOS filter cascade (e.g. in-loop absorption).

One cascade of second-order sections per channel, filtered with scipy.signal.sosfilt(). State persists across filter() calls, so a long signal can be processed in consecutive blocks.

Parameters:: sos (ArrayLike) – Per-channel SOS bank of shape (n_sections, 6, N) where N = num_channels. Section rows are [b0, b1, b2, a0, a1, a2]. This is the canonical SOS bank layout in pyFDN: it matches the FLAMO parallelSOSFilter input and the output of pyFDN.decay_to_first_order_shelf(), pyFDN.decay_to_one_pole(), and pyFDN.decay_to_geq().
Variables:: sos – The same bank in the (N, n_sections, 6) layout scipy.signal.sosfilt() expects, one cascade per row.

Notes

Pass an instance as post_delay to pyFDN.process_fdn() to apply frequency-dependent absorption inside the feedback loop. pyFDN.build_to_impz() also constructs this class internally when an pyFDN.FDNBuild contains per-delay-line filters.

filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

reset()[source]#

Clear internal state (no-op for stateless operators).

Return type:: None

class pyFDN.td.operators.TimeOperator[source]#

Bases: ABC

Abstract class. Parent of all classes belonging to the td-graph group. A stateful (T, in_channels) -> (T, out_channels) time-domain block. Subclasses set in_channels / out_channels and implement filter(). reset() returns the operator to its initial (zero) state.

abstractmethod filter(block)[source]#

Filter one block and advance internal state.

Return type:: ndarray

in_channels: int#

out_channels: int#

process(signal, *, squeeze=False)[source]#

Filter a whole signal in one call, from the current state.

Convenience wrapper around filter() for the common case of rendering an operator tree offline. Operators that process in blocks internally (Recursion) do so regardless of how the signal is handed to them, so this gives the same result as streaming signal through filter() block by block.

Parameters:

signal (ArrayLike) – Input of shape (num_samples,) or (num_samples, in_channels).
squeeze (bool) – Squeeze singleton output channels (default False).

Returns:

Output of shape (num_samples, out_channels).

Return type:

ndarray

reset()[source]#

Clear internal state (no-op for stateless operators).

Return type:: None

class pyFDN.td.operators.TimeVaryingMatrix(N, cycles_per_second, amplitude, fs, spread)[source]#

Bases: TimeOperator

Stateful sinusoidally modulated orthogonal mixing matrix (time-varying feedback).

Each adjacent channel pair is rotated by a sinusoidally modulated angle, so the operator is orthogonal at every sample but never constant. Sitting on a Recursion feedback path, e.g. Series([Gain(A), TimeVaryingMatrix(N, 1.5, 0.35, fs, 0.1)]), it makes the loop genuinely time-varying – there is no static transfer function. This is the operator form of the post_matrix argument of pyFDN.process_fdn().

Translation of the MATLAB implementation timeVaryingMatrix.m from fdnToolbox. Original MATLAB code: (c) Sebastian Jiro Schlecht, 2019. Python translation: Alma Hova, 2026.

Parameters:

N (int) – Number of channels (the matrix is N x N). Must be a positive even integer.
cycles_per_second (float) – Frequency of the time variation in Hz (controls oscillation speed).
amplitude (float) – Maximum angle deflection in radians (strength of modulation).
fs (float) – Sampling rate in Hz.
spread (float) – Randomness factor (controls how differently each eigenmode behaves).

Variables:

num_pairs (int) – Number of eigenmode pairs (N // 2), i.e. independent 2-D rotation planes.
angle_amplitude (phase, frequency,) – Per-pair modulation parameters, drawn from the global NumPy RNG at construction. Seed np.random.seed beforehand for a reproducible modulation.
sample_index (int) – Current sample index; the modulation clock.

filter(block)[source]#

Apply the time-varying orthogonal transformation to one block.

The operation is equivalent to constructing the block-diagonal rotation matrix from rotation_matrix_from_angles at every sample, but applies the 2-D rotations directly to the whole input block.

Return type:: ndarray

reset()[source]#

Clear internal state (no-op for stateless operators).

Return type:: None

Module contents#

Time-domain rendering of block-based processing graphs.

This subpackage builds a processing graph as a tree of stateful NumPy operators and runs it block by block in the time domain – no torch, no FFT. Leaves (Gain, Delay, SOSBank, …) are wired together by the connectors Series, Parallel and Recursion.

Typical use:

import numpy as np
from pyFDN import td

forward = td.Series([td.Delay(delays - block_size), td.SOSBank(absorption)])
fdn = td.Series([
    td.Gain(B),
    td.Recursion(forward, td.Gain(A), block_size=block_size),
    td.Gain(C),
])
ir = fdn.process(impulse)

The feedback Recursion is the only non-trivial piece: it processes in blocks and therefore inserts block_size samples of delay into the loop, which the caller compensates for. See pyFDN.td.connectors.Recursion.