Python FIR Filter API¶

Direct-form FIR filter backed by fir_state_t. Accepts real (float32) or complex (complex64) taps; input must be complex64.

Source: src/doppler/filter/__init__.py

Tap types¶

Tap dtype	C path	Cost/tap/sample	When to use
`float32`	real	1 FMA	`scipy.signal.firwin`, any symmetric LP/HP/BP
`complex64`	complex	2 FMA + permute	Hilbert transformer, frequency-shifted designs

Examples¶

Low-pass filter (real taps)¶

from doppler.filter import FIR
from scipy.signal import firwin
import numpy as np

taps = firwin(63, cutoff=0.1, window="hamming").astype(np.float32)
filt = FIR(taps)

x = np.random.randn(4096).astype(np.complex64)
y = filt.execute(x)     # complex64 out, length 4096

Reusing across blocks (phase-continuous)¶

from doppler.filter import FIR
from scipy.signal import firwin
import numpy as np

taps = firwin(63, cutoff=0.2).astype(np.float32)
filt = FIR(taps)

# a couple of complex64 blocks standing in for a live capture stream
stream = [np.random.randn(256).astype(np.complex64) for _ in range(3)]
for block in stream:                        # generator of complex64 arrays
    out = filt.execute(block)               # state preserved across calls

Complex taps — Hilbert transformer¶

from doppler.filter import FIR
import numpy as np

# Simple 4-tap complex example; use scipy for real designs
ctaps = np.array([0+1j, 0+1j, 0+1j, 0+1j], dtype=np.complex64) / 4
filt = FIR(ctaps)
print(filt.is_real)   # False

Stream discontinuity¶

filt.reset()    # zero delay line; tap coefficients preserved

FIR ¶

Create a FIR filter from complex CF32 tap coefficients. Implements a direct-form FIR convolution: y[n] = sum_k h[k]*x[n-k]. The tap array is copied at creation; the caller may free it afterward. Use fir_create_real() instead when all imaginary parts are zero — that path costs 1 FMA/tap versus 2 FMA + permute + mul here.

Parameters:

Name	Type	Description	Default
`taps`	`NDArray[complex64]`	Array of taps_len CF32 coefficients (I+jQ each), copied.	required

Examples:

>>> import numpy as np
>>> from doppler.filter import FIR
>>> taps = np.array([0.25+0j, 0.5+0j, 0.25+0j], dtype=np.complex64)
>>> fir = FIR(taps)
>>> fir.num_taps
3
>>> fir.is_real
False

num_taps `property` ¶

num_taps: int

Number of tap coefficients supplied at creation. This equals the filter group delay plus one, and determines the minimum input block length for which no latency is observable.

is_real `property` ¶

is_real: bool

True when the filter was created with real-valued tap coefficients. Real-tap filters (fir_create_real) use a cheaper inner loop: 1 FMA/tap versus the 2 FMA + lane permute required for complex multiplication. Use this flag to confirm which constructor path was used at runtime.

reset ¶

reset() -> None

Zero the delay line; preserve taps and scratch capacity. After a reset the filter behaves identically to a freshly constructed instance of the same length, without paying the allocation cost again. Call this between unrelated signal segments to prevent inter-segment leakage through the delay line.

Examples:

>>> import numpy as np
>>> from doppler.filter import FIR
>>> taps = np.array([0.25+0j, 0.5+0j, 0.25+0j], dtype=np.complex64)
>>> fir = FIR(taps)
>>> x = np.array([1+0j, 0+0j, 0+0j], dtype=np.complex64)
>>> _ = fir.execute(x)
>>> fir.reset()
>>> y = fir.execute(x)
>>> [round(float(v.real), 4) for v in y]
[0.25, 0.5, 0.25]

execute ¶

execute(
    x: NDArray[complex64],
    out: NDArray[complex64] | None = None,
) -> NDArray[np.complex64]

Filter n_in CF32 samples and write the results to out. Each output sample is the inner product of the tap vector with the current delay line. The delay line is updated with each input sample so state carries over across successive calls — process frames of any size without gaps or overlap. The scratch buffer is grown lazily on the first call and reused on subsequent calls of the same size.

Parameters:

Name	Type	Description	Default
`x`	`NDArray[complex64]`	Input.	required
`out`	`NDArray[complex64] \| None`	Output buffer; caller must provide space for n_in CF32 values.	`None`

Returns:

Type	Description
`NDArray[complex64]`	Number of output samples written (always == n_in).

Examples:

>>> import numpy as np
>>> from doppler.filter import FIR
>>> taps = np.array([0.25+0j, 0.5+0j, 0.25+0j], dtype=np.complex64)
>>> fir = FIR(taps)
>>> x = np.array([1+0j, 0+0j, 0+0j], dtype=np.complex64)
>>> y = fir.execute(x)
>>> y.dtype
dtype('complex64')
>>> y.shape
(3,)
>>> [round(float(v.real), 4) for v in y]
[0.25, 0.5, 0.25]

execute_max_out ¶

execute_max_out() -> int

Always 0 -- FIR is a 1:1 transform, not a bounded-capacity one.

fir_execute() always writes exactly n_in samples; there is no

call-independent upper bound smaller than the input length for this

function to report. An out= buffer must be sized to exactly

len(x), not to this function's return value.

Returns:

Type	Description
`int`	Output.

state_bytes ¶

state_bytes() -> int

Size in bytes of this object's serialized state.

The exact length get_state returns and set_state requires. It depends on how the object was constructed (state arrays are sized at construction), so read it from the instance rather than assuming a constant.

Raises RuntimeError if the FIR has already been destroyed.

Returns:

Type	Description
`int`	Byte length of one serialized state blob.

get_state ¶

get_state() -> bytes

Serialize this object's mutable state to bytes.

Captures exactly the state that evolves as the object runs, so a blob taken now and restored later resumes from this point. Construction parameters are not included: restore into an object built the same way.

The blob is opaque and always state_bytes() long. Its layout is an implementation detail of the C core and is not a stable format across builds.

Raises RuntimeError if the FIR has already been destroyed.

Returns:

Type	Description
`bytes`	Opaque snapshot, `state_bytes()` bytes long.

set_state ¶

set_state(blob: bytes) -> None

Restore mutable state from a get_state() blob.

Overwrites the live state in place; the object keeps the parameters it was constructed with. Length is validated against state_bytes() before the blob is handed to the C core, and the core may reject it as well.

Raises TypeError if blob is not bytes, ValueError if its length differs from state_bytes() or the core rejects it, and RuntimeError if the FIR has already been destroyed.

Parameters:

Name	Type	Description	Default
`blob`	`bytes`	A `get_state()` blob from this type, exactly `state_bytes()` long.	required

destroy ¶

destroy() -> None

Release the underlying C resources immediately.

Ordinarily unnecessary: the resources are freed when the object is garbage-collected. Call this to release them at a definite point instead, or use the object as a context manager, which calls it on exit.

Idempotent: calling it again on an already-released object does nothing. Every other method raises RuntimeError once it has run.

enter ¶

__enter__() -> FIR

Enter a context manager, returning this object.

Lets a FIR be used in a with statement so its C resources are released deterministically on exit rather than at collection time.

Returns:

Type	Description
`FIR`	This same object, not a copy.

exit ¶

__exit__(
    exc_type: object | None = ...,
    exc: object | None = ...,
    tb: object | None = ...,
) -> None

Exit a context manager, releasing the FIR.

Equivalent to calling destroy(). Returns None, so an exception raised inside the with body propagates normally; this never suppresses one.

Parameters:

Name	Type	Description	Default
`exc_type`	`object \| None`	Exception class, or None. Ignored.	`...`
`exc`	`object \| None`	Exception instance, or None. Ignored.	`...`
`tb`	`object \| None`	Traceback object, or None. Ignored.	`...`

Filter design helpers¶

design_lowpass composes doppler.resample.kaiser_num_taps / kaiser_beta and doppler.spectral.kaiser_window into a one-call windowed-sinc lowpass design: n_taps is sized automatically from the requested passband/stopband edges and stopband attenuation — no hand-rolled sinc/window math or scipy dependency required. fpass/fstop are Nyquist-normalised (1.0 == fs/2), matching the convention kaiser_num_taps already uses.

from doppler.filter import FIR, design_lowpass
import numpy as np

taps = design_lowpass(fpass=0.4, fstop=0.6, atten_db=60.0)
filt = FIR(taps)

x = np.random.randn(4096).astype(np.complex64)
y = filt.execute(x)

design_lowpass ¶

design_lowpass(
    fpass: float = 0.4,
    fstop: float = 0.6,
    atten_db: float = 60.0,
) -> NDArray[np.float32]

Kaiser-windowed-sinc lowpass FIR taps, auto-sized by kaiser_num_taps (Nyquist-normalised fpass/fstop band edges, unit-DC-gain float32 taps).

Moving average (boxcar)¶

MovingAverage is a sliding-window boxcar filter over the last len complex samples — one output per input sample (no rate change). Each step adds the new sample and subtracts the sample leaving the window, so it is O(1) per sample regardless of window length (a running window sum, not a re-summed convolution). The output is the window mean times an optional output gain, folded into a single cached scale = gain/len so applying the gain is free. The delay ring is a fixed in-struct array, so the state is pointer-free POD: it embeds by value into a composing object (a carrier loop's I/Q arm, a smoother ahead of a detector) and serializes as a whole-struct snapshot.

import numpy as np
from doppler.filter import MovingAverage

ma = MovingAverage(2)                       # 2-sample window, unit gain
ma.steps(np.ones(3, np.complex64)).real     # [0.5, 1.0, 1.0] — ramps in

ma2 = MovingAverage(4, gain=2.0)            # gain folded into the mean
y = ma2.step(1.0 + 0.0j)                    # one sample, returns the gained mean

MovingAverage ¶

MovingAverage component.

Parameters:

Name	Type	Description	Default
`len`	`int`	len constructor parameter.	`4`
`gain`	`float`	gain constructor parameter.	`1.0`

Examples:

Create with defaults:

>>> from doppler.filter import MovingAverage
>>> obj = MovingAverage(len=4, gain=1.0)

len `property` ¶

len: int

window length (1 .. BOXCAR_MAX_LEN).

gain `property` `writable` ¶

gain: float

Current output gain.

step ¶

step(x: complex) -> complex

Slide the window by one sample; return the gained moving average.

O(1): add x, drop the sample leaving the window, return acc · scale (= gain · acc / len) — one multiply.

Parameters:

Name	Type	Description	Default
`x`	`complex`	One input sample.	required

Returns:

Type	Description
`complex`	The gained window mean after admitting x.

Examples:

>>> from doppler.filter import MovingAverage
>>> ma = MovingAverage(2)   # 2-sample sliding window, unit gain
>>> [round(ma.step(v).real, 4) for v in (1 + 0j, 3 + 0j, 3 + 0j)]
[0.5, 2.0, 3.0]

steps ¶

steps(
    x: NDArray[complex64],
    out: NDArray[complex64] | None = None,
) -> NDArray[np.complex64]

Filter a block: write the gained moving average of each sample.

Applies boxcar_step() to each input sample in turn, so the window sum and ring carry across the block exactly as they would sample by sample — a stream can be processed in frames of any size with no seam. Immediately after a reset the first len-1 outputs average over a partial (still filling) window and ramp in.

Parameters:

Name	Type	Description	Default
`x`	`NDArray[complex64]`	Input samples.	required

Returns:

Type	Description
`NDArray[complex64]`	Output.

Examples:

>>> import numpy as np
>>> from doppler.filter import MovingAverage
>>> ma = MovingAverage(3)                          # 3-sample window
>>> x = np.ones(5, np.complex64)                   # unit step input
>>> [round(v, 4) for v in ma.steps(x).real.tolist()]
[0.3333, 0.6667, 1.0, 1.0, 1.0]

reset ¶

reset() -> None

Clear the window (zero the ring and the running sum); keep the configured length and gain.

Returns the filter to its just-constructed state: the delay ring and the running window sum are zeroed while len and gain are preserved, so the next len-1 outputs ramp in over a partial window exactly as they did on a fresh instance. Call it at a segment boundary so samples from one capture do not average into an unrelated next one.

Examples:

>>> import numpy as np
>>> from doppler.filter import MovingAverage
>>> ma = MovingAverage(2)                         # 2-sample window
>>> _ = ma.steps(np.ones(4, np.complex64))        # fill the window
>>> ma.reset()                                    # clear it
>>> round(ma.step(1 + 0j).real, 4)                # ramps in from empty
0.5

state_bytes ¶

state_bytes() -> int

Size in bytes of this object's serialized state.

The exact length get_state returns and set_state requires. It depends on how the object was constructed (state arrays are sized at construction), so read it from the instance rather than assuming a constant.

Raises RuntimeError if the MovingAverage has already been destroyed.

Returns:

Type	Description
`int`	Byte length of one serialized state blob.

get_state ¶

get_state() -> bytes

Serialize this object's mutable state to bytes.

Captures exactly the state that evolves as the object runs, so a blob taken now and restored later resumes from this point. Construction parameters are not included: restore into an object built the same way.

The blob is opaque and always state_bytes() long. Its layout is an implementation detail of the C core and is not a stable format across builds.

Raises RuntimeError if the MovingAverage has already been destroyed.

Returns:

Type	Description
`bytes`	Opaque snapshot, `state_bytes()` bytes long.

set_state ¶

set_state(blob: bytes) -> None

Restore mutable state from a get_state() blob.

Overwrites the live state in place; the object keeps the parameters it was constructed with. Length is validated against state_bytes() before the blob is handed to the C core, and the core may reject it as well.

Raises TypeError if blob is not bytes, ValueError if its length differs from state_bytes() or the core rejects it, and RuntimeError if the MovingAverage has already been destroyed.

Parameters:

Name	Type	Description	Default
`blob`	`bytes`	A `get_state()` blob from this type, exactly `state_bytes()` long.	required

destroy ¶

destroy() -> None

Release the underlying C resources immediately.

Ordinarily unnecessary: the resources are freed when the object is garbage-collected. Call this to release them at a definite point instead, or use the object as a context manager, which calls it on exit.

Idempotent: calling it again on an already-released object does nothing. Every other method raises RuntimeError once it has run.

enter ¶

__enter__() -> MovingAverage

Enter a context manager, returning this object.

Lets a MovingAverage be used in a with statement so its C resources are released deterministically on exit rather than at collection time.

Returns:

Type	Description
`MovingAverage`	This same object, not a copy.

exit ¶

__exit__(
    exc_type: object | None = ...,
    exc: object | None = ...,
    tb: object | None = ...,
) -> None

Exit a context manager, releasing the MovingAverage.

Equivalent to calling destroy(). Returns None, so an exception raised inside the with body propagates normally; this never suppresses one.

Parameters:

Name	Type	Description	Default
`exc_type`	`object \| None`	Exception class, or None. Ignored.	`...`
`exc`	`object \| None`	Exception instance, or None. Ignored.	`...`
`tb`	`object \| None`	Traceback object, or None. Ignored.	`...`

Guides — Checkpoint & Resume Design — Design — pure-functional acquisition kernel (elastic fleet), API taxonomy: the DSP building-block hierarchy and its naming axis, State Serialization — the standard bytes interface Contributing — Python Extension Module Layout

Python FIR Filter API¶

Tap types¶

Examples¶

Low-pass filter (real taps)¶

Reusing across blocks (phase-continuous)¶

Complex taps — Hilbert transformer¶

Stream discontinuity¶

FIR ¶

num_taps property ¶

is_real property ¶

reset ¶

execute ¶

execute_max_out ¶

state_bytes ¶

get_state ¶

set_state ¶

destroy ¶

__enter__ ¶

__exit__ ¶

Filter design helpers¶

design_lowpass ¶

Moving average (boxcar)¶

MovingAverage ¶

len property ¶

gain property writable ¶

step ¶

steps ¶

reset ¶

state_bytes ¶

get_state ¶

set_state ¶

destroy ¶

__enter__ ¶

__exit__ ¶

Related pages¶

num_taps `property` ¶

is_real `property` ¶

enter ¶

exit ¶

len `property` ¶

gain `property` `writable` ¶

enter ¶

exit ¶