dspk.ops

Signal-processing operations of dspk-ops, run on the chosen runtime (see dspk.runtime).

One-shot functions (dspk.ops.filter) upload their input, run and return NumPy. A Pipeline chains stage stages on the device: one upload, every stage, one download per chunk, with filter designs and device buffers kept between chunks.

Pipeline	Stages run on the device over chunks of a signal.

Classes

Pipeline

Pipeline(
    stages: Sequence[stage.Stage],
    sample_rate: float,
    *,
    stream: bool = False,
    runtime: str | None = None,
) -> None

Stages run on the device over chunks of a signal: one upload, every stage, one download.

Filter designs and device buffers are kept between chunks.

Raises

• ValueError: When a stage is invalid at sample_rate, or stream=True has a zero-phase filter.

Examples

>>> import numpy as np
>>> import dspk
>>> x = np.random.default_rng(0).standard_normal((4, 30_000)).astype(np.float32)
>>> chain = dspk.ops.Pipeline(
...     [dspk.ops.stage.iir("highpass", 300.0), dspk.ops.stage.cmr()], 30_000.0
... )
>>> chain.process(x).shape
(4, 30000)

Parameters

• stages : sequence of Stage
  The stages, in order (stage).
• sample_rate : float
  Sample rate of the data, in Hz.
• stream : bool, default False
False: every chunk on its own (zero-phase filters allowed; give each chunk settling of context). True: chunks are consecutive pieces of one signal and filter state carries over, so chunked output equals whole-signal output (forward filters only).
• runtime : str, default None
  A runtime from runtime.available ("cuda", "hip", "wgpu", "cpu") for this call. By default: the one chosen with runtime.set, else DSPK_RUNTIME, else the first available (GPUs first).

Properties

Name	Type	Description
settling	tuple[int, int]	Context (left, right) in samples a chunk needs for its output to match the whole signal’s.
sample_rate	float	Sample rate of the data, in Hz.
stream	bool	Whether chunks are consecutive pieces of one signal.
runtime	str	The runtime the stages run on.

Pipeline.process

process(data) -> np.ndarray

Runs every stage over a chunk.

Parameters

• data : array_like
  Signal (channels, samples). float32, int8, int16, uint16 and int32 arrays go to the device as they are and are converted there; other dtypes and lists become float32 first.

Returns

• np.ndarray — (channels, samples), float32.

Raises

• ValueError: When a spatial stage was made for another channel count.
• RuntimeError: When the runtime has no device.

Pipeline.reset

reset() -> None

Forgets the device workspace: the next chunk starts a new stream.