Python Delay Line API

Dual-write circular delay line for complex128 samples, backed by dp_delay_cf64_t. Designed for polyphase FIR resamplers that need a contiguous window of history with no modulo arithmetic.

Source: src/doppler/delay/__init__.py

How it works

The buffer holds 2 × capacity samples, where capacity is the smallest power of two ≥ num_taps. Every push writes the new sample at buf[head] and buf[head + capacity]. ptr() always returns a contiguous num_taps-window — no wrap-around branch needed by the FIR kernel.

Newest sample is at index 0; oldest at index num_taps - 1.

Examples

Basic push and read-back

from doppler.delay import DelayCf64
import numpy as np

dl = DelayCf64(4)           # 4-tap window

dl.push(1+2j)
dl.push(3+4j)

window = dl.ptr()           # array([3+4j, 1+2j, 0+0j, 0+0j])
print(window[0])            # newest: 3+4j

Polyphase FIR inner loop

from doppler.delay import DelayCf64
import numpy as np

num_taps = 19
dl = DelayCf64(num_taps)

taps = np.ones(num_taps, dtype=np.float32) / num_taps   # example taps

for sample in iq_stream:
    dl.push(sample)
    window = dl.ptr()                       # contiguous num_taps window
    out = np.dot(window, taps.astype(np.complex128))

Push and read in one call

push_ptr pushes a sample and returns the updated window in one round-trip to C, saving one Python call per sample.

for sample in iq_stream:
    window = dl.push_ptr(sample)
    out = np.dot(window, taps)

Context manager

with DelayCf64(32) as dl:
    for s in stream:
        dl.push(s)
    # dl released on exit

DelayCf64

Dual-write circular delay line for complex128 samples.

Wraps dp_delay_cf64_t. Keeps the most-recent num_taps samples in a contiguous window so polyphase FIR kernels can MAC directly without modulo arithmetic.

Buffer size is 2 × capacity, where capacity is the smallest power of two ≥ num_taps. :meth:push writes to both halves; :meth:ptr always returns a contiguous window at the current head. Newest sample is at index 0, oldest at index num_taps - 1.

Parameters:

Name	Type	Description	Default
`num_taps`	`int`	Window length (number of samples to retain).	required

Examples:

>>> from doppler.delay import DelayCf64
>>> dl = DelayCf64(4)
>>> dl.push(1+2j)
>>> dl.push(3+4j)
>>> dl.ptr()
array([3.+4.j, 1.+2.j, 0.+0.j, 0.+0.j])

num_taps `property`

num_taps: int

Window length passed at construction.

capacity `property`

capacity: int

Actual allocated capacity (smallest power-of-two ≥ num_taps).

push

push(x: complex) -> None

Push one complex128 sample into the delay line.

push_ptr

push_ptr(x: complex) -> NDArray[np.complex128]

Push one sample and return the updated window in one call.

write

write(x: NDArray[complex128]) -> None

Push all samples from a complex128 array.

ptr

ptr() -> NDArray[np.complex128]

Return a copy of the current num_taps-sample window.

reset

reset() -> None

Zero all samples in the delay line.

destroy

destroy() -> None

Release C resources explicitly.