Batched adaptive lattice trials

Tutorial goal

Run independent adaptive system-identification trials through the batch API.

Note

New to the terminology? See the lattice DSP concept map and the causality/data-use guide for how online, offline, block, and MIMO examples should be read.

Context

Parameter sweeps and Monte Carlo trials are easier when many independent filters can be run together. This example mirrors the scalar adaptive API but uses a batch-oriented entry point.

Key idea and equations

Each batch member has its own input, desired signal, and final state. The jobs are independent, so they can be parallelized safely.

How to read the result

Confirm that each batch member converges and that the reported batch dimensions match the requested number of trials.

Run command

python examples/adaptive_batch_processing.py

Run status

Return code: 0

Captured stdout

initial MSE: 0.02824192672167715
final MSE: 0.0006448983792747414
mean final reflection: [ 0.23467429 -0.2368577 ]
mean final taps: [ 0.49709591 -0.20342426  0.11425937]

Source code

"""Run independent adaptive IIR identification problems in a single C++/OpenMP call."""

import numpy as np

from lattice_dsp import LatticeIIR, adaptive_process_batch

rng = np.random.default_rng(42)
channels = 8
samples = 4000
x = rng.normal(size=(channels, samples))

true_reflection = [0.35, -0.2]
true_taps = [0.5, -0.15, 0.1]

desired = np.vstack(
    [np.asarray(LatticeIIR(true_reflection, true_taps).process(row), dtype=float) for row in x]
)

y, error, final_reflection, final_taps = adaptive_process_batch(
    [0.0, 0.0],
    [0.0, 0.0, 0.0],
    x,
    desired,
    mu_taps=0.08,
    mu_reflection=0.002,
    n_threads=0,
)

print("initial MSE:", np.mean(error[:, :500] ** 2))
print("final MSE:", np.mean(error[:, -500:] ** 2))
print("mean final reflection:", final_reflection.mean(axis=0))
print("mean final taps:", final_taps.mean(axis=0))