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 num_taps CF32 coefficients (I+jQ each), copied.	`...`

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]) -> 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

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]

state_bytes ¶

state_bytes() -> int

Serialized state size in bytes.

get_state ¶

get_state() -> bytes

Serialize the engine's mutable state to bytes.

set_state ¶

set_state(blob: bytes) -> None

Restore mutable state from a get_state() blob.

destroy ¶

destroy() -> None

Release C resources immediately.

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

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.

steps ¶

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

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

Parameters:

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

Returns:

Type	Description
`NDArray[complex64]`	Output.

reset ¶

reset() -> None

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

state_bytes ¶

state_bytes() -> int

Serialized state size in bytes.

get_state ¶

get_state() -> bytes

Serialize the engine's mutable state to bytes.

set_state ¶

set_state(blob: bytes) -> None

Restore mutable state from a get_state() blob.

destroy ¶

destroy() -> None

Release C resources immediately.

Fixed-point halfband decimator (Q15)¶

HBDecimQ15 is a fixed-point halfband 2:1 decimator for interleaved-I/Q int16 streams — the integer-pipeline counterpart to resample.HalfbandDecimator. The FIR branch taps are supplied as float and converted internally to Q15 (with the ×0.5 polyphase rate scaling). The halfband prototype is sparse — every other tap is zero — so you supply only the non-zero branch taps, not the full prototype. See the HBDecimQ15 example for the passband/stopband response.

import numpy as np
from doppler.filter import HBDecimQ15

# non-zero branch taps of a halfband prototype (float; converted to Q15)
taps = np.array([-0.03, 0.28, 0.5, 0.28, -0.03], np.float32)
dec = HBDecimQ15(taps)
x = (np.random.randn(4096) * 8192).astype(np.int16)   # interleaved I/Q
y = dec.execute(x)                                     # int16, half the length

HBDecimQ15 ¶

Allocate and initialise a fixed-point halfband 2:1 decimator. The FIR branch coefficients are supplied as float and converted internally to Q15 with a x0.5 polyphase rate scaling. The full halfband prototype is sparse (every other tap is zero); supply only the non-zero FIR branch taps, not the full sparse prototype.

Parameters:

Name	Type	Description	Default
`h`	`NDArray[float32]`	Float FIR branch coefficients of length num_taps. Must be symmetric (`h[k]` == `h[num_taps-1-k]`).	`...`

num_taps `property` ¶

num_taps: int

FIR branch length as supplied to the constructor. This is the count of non-zero symmetric taps in the FIR branch, not the full sparse halfband prototype length. Useful for introspection when chaining multiple stages with programmatically computed filter banks.

rate `property` ¶

rate: float

The sample-rate reduction factor; always 0.5 for 2:1 decimation. Exposed as a read-only property so pipelines can query the rate of each stage programmatically without hard-coding the 2:1 assumption.

execute ¶

execute(x: NDArray[int16]) -> NDArray[np.int16]

Decimate a block of interleaved IQ int16 samples by 2. Input must be interleaved int16_t IQ pairs (I₀ Q₀ I₁ Q₁ …); pass a 1-D array of 2*n_complex elements. Each pair of complex input samples produces one complex output sample, so an array of length 2N yields at most N output pairs (2N int16 output values). If n_in is odd the trailing IQ pair is buffered and consumed on the next call.

Parameters:

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

Returns:

Type	Description
`NDArray[int16]`	Number of int16_t values written to out.

Examples:

>>> import numpy as np
>>> from doppler.filter import HBDecimQ15
>>> h = np.array([0.25, 0.5, 0.25], dtype=np.float32)
>>> dec = HBDecimQ15(h)
>>> x = np.array([1000, 0, 1000, 0, 1000, 0, 1000, 0], dtype=np.int16)
>>> y = dec.execute(x)
>>> y.dtype
dtype('int16')
>>> y.shape
(4,)
>>> y.tolist()
[0, 0, 625, 0]

reset ¶

reset() -> None

Zero all delay rings and clear the pending-sample flag. After a reset the decimator behaves identically to a freshly constructed instance: the four dual-write delay rings are zeroed and has_pending is cleared, so no partial IQ pair carries over. Call this between unrelated signal segments to prevent inter-segment leakage.

Examples:

>>> import numpy as np
>>> from doppler.filter import HBDecimQ15
>>> h = np.array([0.25, 0.5, 0.25], dtype=np.float32)
>>> dec = HBDecimQ15(h)
>>> x = np.array([1000, 0, 1000, 0, 1000, 0, 1000, 0], dtype=np.int16)
>>> _ = dec.execute(x)
>>> dec.reset()
>>> y = dec.execute(x)
>>> y.tolist()
[0, 0, 625, 0]

state_bytes ¶

state_bytes() -> int

Serialized state size in bytes.

get_state ¶

get_state() -> bytes

Serialize the engine's mutable state to bytes.

set_state ¶

set_state(blob: bytes) -> None

Restore mutable state from a get_state() blob.

destroy ¶

destroy() -> None

Release C resources immediately.

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 ¶

state_bytes ¶

get_state ¶

set_state ¶

destroy ¶

Moving average (boxcar)¶

MovingAverage ¶

len property ¶

gain property writable ¶

step ¶

steps ¶

reset ¶

state_bytes ¶

get_state ¶

set_state ¶

destroy ¶

Fixed-point halfband decimator (Q15)¶

HBDecimQ15 ¶

num_taps property ¶

rate property ¶

execute ¶

reset ¶

state_bytes ¶

get_state ¶

set_state ¶

destroy ¶

num_taps `property` ¶

is_real `property` ¶

len `property` ¶

gain `property` `writable` ¶

num_taps `property` ¶

rate `property` ¶