Skip to content

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

iq_stream = (np.random.randn(64)
             + 1j * np.random.randn(64)).astype(np.complex128)
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

stream = (np.random.randn(64)
          + 1j * np.random.randn(64)).astype(np.complex128)
with DelayCf64(32) as dl:
    for s in stream:
        dl.push(s)
    # dl released on exit

DelayCf64

Create a dual-buffer circular delay line of length num_taps. The internal capacity is rounded up to the next power of two so that modular indexing reduces to a single bitwise AND. Any window of num_taps consecutive samples is always contiguous in the backing store; no wrap-around copy is ever needed.

Parameters:

Name Type Description Default
num_taps int

Number of delay taps (window length, >= 1). Internally rounded up to the next power of two.

 1

Examples:

Create with defaults:

>>> from doppler.delay import DelayCf64
>>> obj = DelayCf64(num_taps=1)

num_taps property 

num_taps: int

Num taps.

capacity property 

capacity: int

Capacity.

reset 

reset() -> None

Reset the delay line to its post-create state. Zeroes the entire dual buffer and resets the write pointer to 0, discarding all previously pushed samples. The num_taps and capacity are preserved; only the sample history is cleared.

Examples:

>>> from doppler.delay import DelayCf64
>>> d = DelayCf64(num_taps=3)
>>> d.push(1+2j)
>>> d.push(3+4j)
>>> d.ptr().tolist()
[(3+4j), (1+2j), 0j]
>>> d.reset()
>>> d.ptr().tolist()
[0j, 0j, 0j]

push 

push(x: complex) -> None

Advance the write pointer and insert a new sample. The head pointer decrements (mod capacity) before the write so that buf[head] always holds the most recent sample. The same value is simultaneously written at buf[head + capacity] to keep the mirror half in sync; this ensures any num_taps-length window starting at head is contiguous without an extra copy.

Parameters:

Name Type Description Default
x complex

New complex sample to insert.

 required

Examples:

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

ptr 

ptr(n: int = ..., out: NDArray[complex128] | None = ...) -> NDArray[np.complex128]

Return a zero-copy view of the n most recent samples. Copies at most min(n, num_taps) samples starting from buf[head] into out. Because the dual-buffer layout guarantees contiguity, this is a single memcpy of up to num_taps elements; no wrap-around logic is needed. Without out=, the Python binding returns a NumPy array backed directly by the pre-allocated output buffer (base object is the DelayCf64 itself); with out= (must have at least max(ptr_max_out(), n) elements), writes directly into the caller's array and returns a view of it.

Parameters:

Name Type Description Default
n int

Number of most recent samples to return.

 1
out NDArray[complex128]

Caller-provided output buffer.

 ...

Returns:

Type Description
NDArray[complex128]

Number of samples written.

Examples:

>>> from doppler.delay import DelayCf64
>>> d = DelayCf64(num_taps=3)
>>> d.push(1+0j)
>>> d.push(2+0j)
>>> y = d.ptr()
>>> y.tolist()
[(2+0j), (1+0j), 0j]
>>> y.dtype
dtype('complex128')
>>> y.shape
(3,)

ptr_max_out 

ptr_max_out() -> int

Max output length ptr() can produce for the current state. Use to size the out= buffer.

push_ptr 

push_ptr(x: complex, out: NDArray[complex128] | None = ...) -> NDArray[np.complex128]

Atomically push a sample and snapshot the current window. Equivalent to calling push(x) then ptr(num_taps), but avoids the overhead of a second function call. Always writes exactly num_taps samples. Without out=, the Python binding returns a NumPy array backed by the pre-allocated push_ptr output buffer; with out= (must have exactly num_taps elements), writes directly into the caller's array and returns it.

Parameters:

Name Type Description Default
x complex

New complex sample to insert.

 required
out NDArray[complex128]

Caller-provided output buffer; must have exactly num_taps elements.

 ...

Returns:

Type Description
NDArray[complex128]

num_taps (always equal to the window length).

Examples:

>>> from doppler.delay import DelayCf64
>>> d = DelayCf64(num_taps=3)
>>> d.push_ptr(1+0j).tolist()
[(1+0j), 0j, 0j]
>>> d.push_ptr(2+0j).tolist()
[(2+0j), (1+0j), 0j]

push_ptr_max_out 

push_ptr_max_out() -> int

Max output length push_ptr() can produce for the current state (always exactly num_taps). Use to size the out= buffer.

write 

write(x: complex) -> None

Alias for delay_push(); insert a sample without reading back. Provided for API symmetry with write-then-read patterns where the caller wants to decouple sample ingestion from window inspection. Internally delegates to delay_push() with no additional overhead.

Parameters:

Name Type Description Default
x complex

New complex sample to insert.

 required

Examples:

>>> from doppler.delay import DelayCf64
>>> d = DelayCf64(num_taps=2)
>>> d.write(5+6j)
>>> d.ptr().tolist()
[(5+6j), 0j]

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.