Skip to content

Python FIR Filter API

Direct-form FIR filter with AVX-512 acceleration, backed by dp_fir_*. Accepts real or complex taps; input dtype is auto-dispatched to the correct C hot path.

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

Input dtype dispatch

execute(x) routes to the right C kernel based on x.dtype:

Input dtype Interpretation Output
complex64 CF32 IQ complex64
int8 CI8 interleaved I/Q pairs complex64
int16 CI16 interleaved I/Q pairs complex64
int32 CI32 interleaved I/Q pairs complex64

Integer inputs must have an even number of elements; each (I, Q) pair becomes one complex sample with no scaling.

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)

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

SDR front-end: CI16 raw IQ from ADC

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

taps = firwin(31, 0.05).astype(np.float32)
filt = FIR(taps)

# raw_iq: int16 array of interleaved [I0, Q0, I1, Q1, ...]
raw_iq = np.frombuffer(adc_buffer, dtype=np.int16)
y = filt.execute(raw_iq)    # CI16 → CF32 in one call

Stream discontinuity

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

FIR

Direct-form FIR filter.

Parameters:

Name Type Description Default
taps ArrayLike

Filter coefficients as a 1-D numpy array. Accepted dtypes:

  • float32 — real-tap filter (one FMA/tap/output, halves the multiply count relative to the complex path).
  • complex64 — complex-tap filter (full complex multiply).

Other array-likes are cast to complex64 if possible.

 required

Examples:

Low-pass filter with scipy-designed taps (real, float32):

>>> import numpy as np
>>> from doppler.filter import FIR
>>> taps = np.array([0.25, 0.25, 0.25, 0.25], dtype=np.float32)
>>> fir = FIR(taps)
>>> fir.num_taps
4
>>> fir.is_real
True
>>> x = np.ones(8, dtype=np.complex64)
>>> y = fir.execute(x)
>>> y.dtype
dtype('complex64')
>>> float(y[-1].real)
1.0

Hilbert-transformer (complex taps):

>>> ctaps = np.array([1+0j, 0+1j], dtype=np.complex64)
>>> cfir = FIR(ctaps)
>>> cfir.is_real
False

num_taps property 

num_taps: int

Number of tap coefficients.

is_real property 

is_real: bool

True if the filter was created with real (float32) taps.

execute 

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

Filter a block of samples.

Parameters:

Name Type Description Default
x ArrayLike

Input samples. Accepted dtypes:

  • complex64 — n complex samples.
  • int8, int16, int32 — interleaved I/Q; must have an even number of elements; produces n // 2 output samples.
 required

Returns:

Type Description
NDArray[complex64]

Filtered output. Length equals the number of complex input samples. This is a zero-copy view into an internal buffer that is valid until the next execute call.

Examples:

>>> import numpy as np
>>> from doppler.filter import FIR
>>> taps = np.ones(4, dtype=np.float32) / 4
>>> fir = FIR(taps)
>>> x = (np.arange(8) + 1j * np.arange(8)).astype(np.complex64)
>>> y = fir.execute(x)
>>> y.shape
(8,)
>>> y.dtype
dtype('complex64')

reset 

reset() -> None

Zero the delay line without freeing the filter.

Use after a stream discontinuity to prevent history contamination. Tap coefficients and pre-allocated buffers are preserved.

Examples:

>>> import numpy as np
>>> from doppler.filter import FIR
>>> taps = np.array([0.5, 0.5], dtype=np.float32)
>>> fir = FIR(taps)
>>> _ = fir.execute(np.ones(4, dtype=np.complex64))
>>> fir.reset()
>>> y = fir.execute(np.ones(1, dtype=np.complex64))
>>> float(y[0].real)
0.5