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:

>>> from doppler.delay import DelayCf64
>>> d = DelayCf64(num_taps=3)
>>> d.num_taps
3
>>> d.capacity   # next power-of-two >= 3
4

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(
    count: int = ..., out: NDArray[complex128] | None = 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. The Python binding returns a NumPy array backed directly by the pre-allocated output buffer (base object is the DelayCf64 itself).

Parameters:

Name	Type	Description	Default
`count`	`int`	How many output samples to ask for. The call may return fewer; size an `out=` buffer with the matching `_max_out()` when you need the worst case.	`...`
`out`	`NDArray[complex128] \| None`	Output buffer; must hold at least max_out elements.	`None`

Returns:

Type	Description
`NDArray[complex128]`	min(n, num_taps, max_out) samples.

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(n: int) -> int

Maximum samples delay_ptr() writes for a request of n. Returns min(n, num_taps) — the tight per-call bound (gh-607).

Parameters:

Name	Type	Description	Default
`n`	`int`	Number of samples the matching delay_ptr() call requests.	required

Returns:

Type	Description
`int`	min(n, num_taps).

push_ptr ¶

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

Atomically push a sample and snapshot the current window. Equivalent to calling delay_push() then delay_ptr(num_taps), but avoids the overhead of a second function call. Always writes exactly num_taps samples to out. The Python binding returns a NumPy array backed by the pre-allocated push_ptr output buffer.

Parameters:

Name	Type	Description	Default
`x`	`complex`	New complex sample to insert.	required
`out`	`NDArray[complex128] \| None`	Output buffer; must hold at least max_out elements.	`None`

Returns:

Type	Description
`NDArray[complex128]`	min(num_taps, max_out) samples.

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

Return the maximum output capacity for delay_push_ptr(). Returns num_taps; the Python binding uses this to pre-allocate the output buffer before calling delay_push_ptr().

Returns:

Type	Description
`int`	num_taps (number of samples delay_push_ptr() will write).

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

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 DelayCf64 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 DelayCf64 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 DelayCf64 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__() -> DelayCf64

Enter a context manager, returning this object.

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

Returns:

Type	Description
`DelayCf64`	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 DelayCf64.

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.	`...`

Design — API taxonomy: the DSP building-block hierarchy and its naming axis

Python Delay Line API¶

How it works¶

Examples¶

Basic push and read-back¶

Polyphase FIR inner loop¶

Push and read in one call¶

Context manager¶

DelayCf64 ¶

num_taps property ¶

capacity property ¶

reset ¶

push ¶

ptr ¶

ptr_max_out ¶

push_ptr ¶

push_ptr_max_out ¶

write ¶

state_bytes ¶

get_state ¶

set_state ¶

destroy ¶

__enter__ ¶

__exit__ ¶

Related pages¶

num_taps `property` ¶

capacity `property` ¶

enter ¶

exit ¶