Streaming: pieces equal the whole

A live recording arrives in pieces. With Pipeline(..., stream=True) each filter carries its state from one piece to the next, so the pieces, of any lengths, give exactly the result of the whole signal at once. Streaming needs causal filters (mode="forward"): a zero-phase filter would need samples that have not arrived yet.

import numpy as np

from dspk.ops import Pipeline, stage
from plating import signals

s = signals.synthetic()
stages = [stage.notch(60.0, 30.0, mode="forward"), stage.iir("bandpass", (300.0, 6_000.0), mode="forward"), stage.cmr()]

whole = Pipeline(stages, s.sample_rate, stream=True).process(s.data)
edges = [0, 1_234, 1_235, 20_000, 33_333, 47_000, s.samples]  # uneven, one of a single sample
stream = Pipeline(stages, s.sample_rate, stream=True)
pieces = np.concatenate([stream.process(s.data[:, a:b]) for a, b in zip(edges, edges[1:])], axis=1)
err = np.max(np.abs(pieces - whole)) / np.max(np.abs(whole))
print(f"{len(edges) - 1} pieces vs the whole signal: max difference {err:.1e} of the peak")

The whole script is playground/06_pipelines/02_streaming.py. Run it with uv run python playground/06_pipelines/02_streaming.py.