Python Ring Buffer API¶

Lock-free SPSC ring buffers backed by dp_buffer_*. Uses virtual-memory double-mapping so the consumer always sees a contiguous window across the wrap boundary — zero-copy, branch-free.

Source: src/doppler/buffer/__init__.py

Buffer types¶

Class	NumPy dtype	Bytes/sample
`F32Buffer`	`complex64`	8
`F64Buffer`	`complex128`	16
`I16Buffer`	`[('i','<i2'),('q','<i2')]`	4

Any capacity from 1 up, and capacity is exactly the number you passed — on every machine. It need not be a power of two.

What is rounded is the mapping behind the ring: up to a power of two, because indexing is a mask, and up to a whole number of pages, because the double mapping is built from them (4 KiB on Linux x86-64, 16 KiB on macOS arm64, 64 KiB allocation granularity on Windows). That is address space, not room — a ring of 1,000 holds 1,000 and refuses the next sample — and it costs nothing per call. A capacity just past a power of two maps nearly twice what it holds; one that is a power of two and spans a page maps exactly itself.

Two surfaces, one ring¶

You are	Read with	Write with
a consumer thread, producer apart	`wait(n)` — blocks	`write(arr)`
one thread doing both	`peek(n)` — never	`write_some(arr)`

Threaded. One producer thread calls write; one consumer thread calls wait / consume. write is non-blocking and refuses the whole call if the ring has no room — it copies nothing and leaves your array untouched, so you still hold the data and can retry. Nothing is dropped unless you discard it; dropped counts refused calls, not lost samples. wait blocks the consumer and releases the GIL so the producer can run concurrently.

Single-threaded. wait would deadlock a caller that is its own producer, so peek(n) returns the same zero-copy view when n samples are there and None when they are not yet. None means not yet and nothing else: a closed ring with fewer than n left raises EOFError, and n > capacity raises ValueError, exactly as wait does. write_some takes what fits and returns the count, never refusing and never touching dropped — the only way to feed a chunk larger than the ring. space is the room a write is guaranteed to find; reset() empties and reopens the ring (both sides must be idle).

Both surfaces release with consume(k); k < n keeps the overlap, which is how overlapped frames are read.

Examples¶

Producer / consumer (threaded)¶

From src/doppler/examples/buffers_demo.py. write() never blocks, so the producer waits for space; close() is how the consumer learns there is no more, and it arrives as EOFError once the ring is drained.

import threading

import numpy as np

from doppler.buffer import F32Buffer, F64Buffer, I16Buffer

IQ16 = np.dtype([("i", "<i2"), ("q", "<i2")])


# A ramp in each width's own element, so a sample says where it came from.
def ramp_complex(dtype):
    return lambda n: np.arange(n).astype(dtype)


def ramp_iq16(n):
    x = np.zeros(n, dtype=IQ16)
    x["i"] = np.arange(n)
    return x


# Two threads. wait() releases the GIL while it spins, so the producer
# runs; close() is how the consumer learns there is no more.
buf = F64Buffer(4096)
got = []


def producer():
    for k in range(8):
        while buf.space < 512:  # write() never blocks: wait for room
            pass
        buf.write(np.full(512, k, dtype=np.complex128))
    buf.close()


t = threading.Thread(target=producer)
t.start()
try:
    while True:
        got.append(int(buf.wait(512)[0].real))
        buf.consume()
except EOFError:
    pass  # closed and drained: the normal end of a stream
t.join()

assert got == list(range(8)), "every block, once, in order"
assert buf.dropped == 0
buf.destroy()

One thread: any block in, exact frames out¶

wait(n) blocks until a producer on another thread delivers, so a caller that is its own producer would wait forever. peek(n) is the same zero-copy view without the wait — the frame, or None for "not yet" — and write_some takes what fits. Together they are the whole single-threaded pattern: feed, take every whole frame, repeat until the block is gone.

import threading

import numpy as np

from doppler.buffer import F32Buffer
from doppler.spectral import FFT


# A ramp, so every sample says where in the stream it came from: a frame
# that is off by one, duplicated or dropped shows up as a break in the
# sequence rather than as a plausible-looking block of signal.
def stream(start: int, n: int) -> np.ndarray:
    idx = np.arange(start, start + n)
    return (idx + 1j * idx).astype(np.complex64)


FFT_N = 1024
ring = F32Buffer(4096)  # four frames. SMALLER than the largest block below.

# What the device hands over: nothing is a multiple of FFT_N, and two of
# them are larger than the whole ring.
DRIVER_BLOCKS = [3000, 5000, 1700, 4096, 777, 2048, 6000, 1234]

fft = FFT(FFT_N, -1)
spectra: list[np.ndarray] = []
frames: list[np.ndarray] = []
produced = 0

for _ in range(6):
    for n in DRIVER_BLOCKS:
        block = stream(produced, n)
        produced += n

        # THE PATTERN. Feed what fits, take every whole frame, repeat until
        # the block is gone. `write_some` never refuses, so there is no
        # room to make first and no block too large; `peek` never blocks,
        # so one thread can do both jobs.
        fed = 0
        while fed < len(block):
            fed += ring.write_some(block[fed:])
            while (frame := ring.peek(FFT_N)) is not None:
                spectra.append(fft.execute_cf32(frame))  # straight in: no copy
                frames.append(frame.copy())  # kept past consume(): copied
                ring.consume()  # releases the frame that was lent

One thread: a chunk larger than the ring, overlapped frames out¶

From src/doppler/examples/ring_nonblocking_demo.py, which checks every frame against the input:

import numpy as np

from doppler.buffer import F32Buffer


def stream(start: int, n: int) -> np.ndarray:
    idx = np.arange(start, start + n)
    return (idx + 1j * idx).astype(np.complex64)


ring = F32Buffer(4096)
CAP = ring.capacity

ring.reset()
NFFT, HOP = 1000, 250
assert CAP % NFFT, "frames must straddle the wrap for this to show anything"
chunk = stream(0, 5 * CAP)
assert not ring.write(chunk[: CAP + 1]), "write() refuses what cannot fit"
refused = ring.dropped

fed = taken = 0
while fed < len(chunk):
    k = ring.write_some(chunk[fed:])
    fed += k
    drained = 0
    while (frame := ring.peek(NFFT)) is not None:
        assert frame.flags["C_CONTIGUOUS"], "the double-mapping's promise"
        assert np.array_equal(frame, chunk[taken * HOP :][:NFFT]), (
            f"frame {taken} is not the stream at hop {taken * HOP}"
        )
        ring.consume(HOP)  # release a hop, keep the overlap
        taken += 1
        drained += 1
    # A full ring holds a whole frame, so one side always moves. Without
    # this, a regression in either call is a hang rather than a failure.
    assert k or drained, "neither side made progress"

End of stream, then the same ring again¶

ring.close()
tail = ring.available
assert 0 < tail < NFFT
try:
    ring.peek(NFFT)
    raise AssertionError("a closed ring answered 'not yet'")
except EOFError:
    pass  # the rest of that frame is never coming: not None, an error
last = ring.peek(tail)  # what IS there can still be read
assert np.array_equal(last, chunk[-tail:])
ring.consume()

ring.reset()
assert not ring.closed and ring.available == 0 and ring.space == CAP
assert ring.write_some(stream(0, 8)) == 8
assert np.array_equal(ring.peek(8), stream(0, 8))

I16Buffer — raw ADC samples¶

numpy has no complex-integer dtype, so one q15 sample is a record, [("i", "<i2"), ("q", "<i2")]. Both faces speak it, which keeps the ring 1-D with one element per sample like its float siblings. The storage underneath is still interleaved int16, so .view() converts either way with no copy.

from doppler.buffer import I16Buffer
import numpy as np

IQ16 = np.dtype([("i", "<i2"), ("q", "<i2")])

buf = I16Buffer(4096)
adc_bytes = np.zeros(2048 * 2, dtype=np.int16).tobytes()   # ADC byte stream
raw = np.frombuffer(adc_bytes, dtype=IQ16)                  # zero-copy
buf.write(raw)

view = buf.wait(1024)       # shape (1024,), dtype IQ16
I = view["i"]               # strided int16 views, no copy
Q = view["q"]
flat = view.view(np.int16)  # interleaved I, Q, I, Q, ... — also no copy
buf.consume()

A bare int16 array — flat or (n, 2) — is refused with TypeError: it is not an array of samples. And a record refuses arithmetic (view + 1 raises) where a packed int32 would carry across the I/Q boundary and corrupt I silently, which is why it is a record.

A full ring¶

A full ring is not an error: write() refuses the block and returns False — you still hold it — and dropped counts what was refused.

buf = F32Buffer(1000)          # any size; not a power of two here
cap = buf.capacity            # 1000

buf.write(np.ones(cap, dtype=np.complex64))        # True
buf.write(np.ones(1, dtype=np.complex64))          # False: no room
buf.dropped                                         # 1 refused, none lost

The full walk through every refusal is src/doppler/examples/ring_refusals_demo.py; the rest of the suite is indexed on Ring Buffers.

F32Buffer ¶

Lock-free SPSC ring buffer for complex64 (CF32) samples.

Parameters:

Name	Type	Description	Default
`capacity`	`int`	How many samples the ring holds: any size from 1 up, and :attr:`capacity` is exactly this number on every machine. What is rounded is the MAPPING behind it -- up to a power of two, because indexing is a mask, and up to a whole page -- so a capacity that is not a power of two costs some address space (under 2x) and nothing per call.	required

Raises:

Type	Description
`ValueError`	If construction fails. The exception message is `capacity must be at least 1, and small enough to map`.

Examples:

>>> from doppler.buffer import F32Buffer
>>> import numpy as np
>>> buf = F32Buffer(1024)
>>> buf.capacity
1024
>>> buf.write(np.ones(512, dtype=np.complex64))
True

capacity `property` ¶

capacity: int

Buffer capacity in complex samples.

available `property` ¶

available: int

Samples written but not yet consumed.

space `property` ¶

space: int

Free room in samples: the largest :meth:write sure to fit.

dropped `property` ¶

dropped: int

Cumulative samples in REFUSED writes -- not samples lost.

closed `property` ¶

closed: bool

True once the producer has called :meth:close.

write ¶

write(x: NDArray[complex64]) -> bool

Write samples into the buffer without blocking.

Copies the complex64 array into the ring buffer in a single memcpy. If there is not enough free space for all len(x) samples the call is refused entirely — nothing is copied and x is untouched, so you still hold every sample and may retry once the consumer has made room. Nothing is dropped unless you discard it; the refusal is counted in :attr:dropped, which is not a loss count. The array must be 1-D and C-contiguous.

Parameters:

Name	Type	Description	Default
`x`	`NDArray[complex64]`	Samples to write. Must be 1-D and C-contiguous.	required

Returns:

Type	Description
`bool`	`True` if all samples were written; `False` if the ring had no room and the call was refused (`x` untouched).

Examples:

>>> from doppler.buffer import F32Buffer
>>> import numpy as np
>>> buf = F32Buffer(1024)
>>> buf.write(np.array([1+2j, 3+4j], dtype=np.complex64))
True
>>> buf2 = F32Buffer(1024)
>>> buf2.write(np.zeros(1024, dtype=np.complex64))
True
>>> buf2.write(np.zeros(1, dtype=np.complex64))
False

write_some ¶

write_some(x: NDArray[complex64]) -> int

Write as much of x as fits and say how much that was.

The partial-write twin of :meth:write. Where :meth:write refuses a block that does not fit whole, this takes the leading samples that do and returns their count -- 0 when the ring is full. It never refuses, so it never touches :attr:dropped. It is the only way to feed a chunk larger than the ring: loop, advancing by the return value, draining in between.

Parameters:

Name	Type	Description	Default
`x`	`NDArray[complex64]`	Samples to write. Must be 1-D and C-contiguous.	required

Returns:

Type	Description
`int`	Samples accepted, `0 <= k <= len(x)`. The caller still owns `x[k:]`.

Examples:

A chunk three times the size of the ring, fed by looping:

>>> from doppler.buffer import F32Buffer
>>> import numpy as np
>>> buf = F32Buffer(1024)
>>> cap = buf.capacity
>>> chunk = np.ones(3 * cap, dtype=np.complex64)
>>> fed = 0
>>> while fed < len(chunk):
...     fed += buf.write_some(chunk[fed:])
...     _ = buf.peek(buf.available); buf.consume()
>>> fed == 3 * cap, buf.dropped
(True, 0)

wait ¶

wait(n: int) -> NDArray[np.complex64]

Block until n samples are available, then lend a zero-copy view.

Spins (releasing the GIL so a producer thread can run concurrently) until at least n samples have been written by the producer. Returns a 1-D complex64 NumPy array that is a direct view into the double-mapped ring buffer — no data is copied. Because of the double-mapping, the view is always contiguous even when the requested range wraps around the physical end of the ring.

The caller must call :meth:consume before the next call to wait. Using the returned array after consume is undefined behaviour; the producer may overwrite it at any time.

Parameters:

Name	Type	Description	Default
`n`	`int`	Number of complex samples to wait for. Must be positive and not larger than :attr:`capacity`.	required

Returns:

Type	Description
`NDArray[complex64]`	Zero-copy view of the next `n` samples in the ring.

Raises:

Type	Description
`EOFError`	The producer called :meth:`close` and fewer than `n` samples remain. The tail is drained and no more is coming, so the wait ends rather than blocking forever.
`KeyboardInterrupt`	Somebody asked this process to stop, through a :class:`doppler.interrupt.Interrupt` guard -- from any module: the flag is process-wide. Without a guard the spin checks for no signals at all.

Examples:

>>> from doppler.buffer import F32Buffer
>>> import numpy as np
>>> buf = F32Buffer(1024)
>>> buf.write(np.array([1+2j, 3+4j, 5+6j], dtype=np.complex64))
True
>>> view = buf.wait(3)
>>> view.dtype
dtype('complex64')
>>> view.shape
(3,)
>>> view.tolist()
[(1+2j), (3+4j), (5+6j)]
>>> buf.consume(3)

peek ¶

peek(n: int) -> NDArray[np.complex64] | None

:meth:wait that never blocks: a view, or None for not yet.

The single-threaded consumer's read. :meth:wait spins until a producer on another thread delivers, so a caller that is its own producer would deadlock in it; peek answers at once instead. When n samples are buffered it returns the same zero-copy, always-contiguous view :meth:wait would (1-D complex64); otherwise it returns None.

None means not yet and nothing else. The two conditions no amount of waiting can cure are raised, exactly as :meth:wait raises them, so a poll loop cannot mistake either for a slow producer.

Peeking does not consume. Follow it with :meth:consume; a consume(k) with k < n advances by a hop smaller than the frame, which is how overlapped frames are read.

Parameters:

Name	Type	Description	Default
`n`	`int`	Number of samples wanted. Must be positive and not larger than :attr:`capacity`.	required

Returns:

Type	Description
`NDArray[complex64] \| None`	Zero-copy view of the next `n` samples, or `None` when fewer than `n` have been written so far. None when the C call lends nothing (returns NULL) and its status has no exception row.

Raises:

Type	Description
`EOFError`	The ring is closed and fewer than `n` samples remain: the rest is never coming.
`ValueError`	`n` exceeds :attr:`capacity` (or is not positive), so no producer could ever satisfy it.

Examples:

>>> from doppler.buffer import F32Buffer
>>> import numpy as np
>>> buf = F32Buffer(1024)
>>> buf.peek(4) is None
True
>>> buf.write_some(np.ones(8, dtype=np.complex64))
8
>>> buf.peek(4).shape
(4,)
>>> buf.consume(2)
>>> buf.available
6
>>> buf.close()
>>> buf.peek(8)
Traceback (most recent call last):
    ...
EOFError: end of stream: the producer closed the ring

consume ¶

consume(n: int = ...) -> None

Release n samples back to the producer.

Advances the consumer tail pointer by n, making that space available for the producer to overwrite, and ends the loan: the view a :meth:wait or :meth:peek lent must not be used afterwards. If n is omitted it is the count of that outstanding view, so the number is written once. n smaller than the view is how overlapped frames are read: release a hop, keep the rest.

Parameters:

Name	Type	Description	Default
`n`	`int`	Number of samples to release. Defaults to the count of the outstanding :meth:`wait` / :meth:`peek` view.	`...`

Raises:

Type	Description
`ValueError`	`n` exceeds :attr:`available`. Nothing is released: past that point the ring's counts would stop describing it.
`RuntimeError`	`n` was omitted and nothing is outstanding -- no view was lent since the last release, so there is no count to default to.
`ValueError`	If the C call returns a non-zero status. The exception message is `consume(n): n exceeds the samples available; nothing was released`, with the return code appended (gh-869).

Examples:

>>> from doppler.buffer import F32Buffer
>>> import numpy as np
>>> buf = F32Buffer(1024)
>>> buf.write(np.ones(4, dtype=np.complex64))
True
>>> _ = buf.wait(4)
>>> buf.consume()

close ¶

close() -> None

Say that no more data is coming.

The producer's half of end of stream. Until this exists a consumer cannot tell a slow producer from a finished one -- both look like an empty ring -- so :meth:wait had nothing to do but spin. Call it once, after the last write.

Release ordering: every sample written before this is visible to a consumer that observes the flag. Closing does not discard what was already written; :meth:wait keeps returning batches until the ring is drained, and only then raises EOFError.

See docs/design/io-termination.md for the one termination contract shared with the network and disk transports.

Examples:

>>> import numpy as np
>>> from doppler.buffer import F32Buffer
>>> buf = F32Buffer(1024)
>>> buf.close()
>>> buf.closed
True
>>> buf.wait(4)
Traceback (most recent call last):
    ...
EOFError: end of stream: the producer closed the ring

reset ¶

reset() -> None

Empty the ring and reopen it.

Discards everything buffered, and clears :attr:closed so the same ring can carry a second stream -- without it, reuse after :meth:close means destroying and re-mapping. :attr:dropped is a lifetime count and is kept.

Not safe against a concurrent producer or consumer: it moves both ends of the ring. Call it only when both sides are idle.

Examples:

>>> from doppler.buffer import F32Buffer
>>> import numpy as np
>>> buf = F32Buffer(1024)
>>> buf.write_some(np.ones(8, dtype=np.complex64))
8
>>> buf.close()
>>> buf.reset()
>>> buf.available, buf.closed
(0, False)

destroy ¶

destroy() -> None

Close the object; the memory goes with its last user.

Ordinarily unnecessary: the resources are freed when the object is garbage-collected. Call this, or use the object as a context manager, to end its use at a definite point.

Views this object handed out keep the memory they point into, as does a call still running on another thread: it is freed when the last of them, and this object, are gone, not here (gh-2187). Idempotent: calling it again does nothing. Every other method raises RuntimeError once it has run.

enter ¶

__enter__() -> F32Buffer

Enter a context manager, returning this object.

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

Returns:

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

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

F64Buffer ¶

Lock-free SPSC ring buffer for complex128 (CF64) samples.

Parameters:

Name	Type	Description	Default
`capacity`	`int`	How many samples the ring holds: any size from 1 up, and :attr:`capacity` is exactly this number on every machine. What is rounded is the MAPPING behind it -- up to a power of two, because indexing is a mask, and up to a whole page -- so a capacity that is not a power of two costs some address space (under 2x) and nothing per call.	required

Raises:

Type	Description
`ValueError`	If construction fails. The exception message is `capacity must be at least 1, and small enough to map`.

Examples:

>>> from doppler.buffer import F64Buffer
>>> import numpy as np
>>> buf = F64Buffer(512)
>>> buf.capacity >= 512
True
>>> buf.write(np.ones(256, dtype=np.complex128))
True

capacity `property` ¶

capacity: int

Buffer capacity in complex samples.

available `property` ¶

available: int

Samples written but not yet consumed.

space `property` ¶

space: int

Free room in samples: the largest :meth:write sure to fit.

dropped `property` ¶

dropped: int

Cumulative samples in REFUSED writes -- not samples lost.

closed `property` ¶

closed: bool

True once the producer has called :meth:close.

write ¶

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

Write complex128 samples into the buffer without blocking.

Copies the entire array in a single memcpy. Rejects the write atomically if there is insufficient free space; the call is refused whole -- nothing copied, x untouched -- and :attr:dropped grows by len(x). The array must be 1-D and C-contiguous.

Parameters:

Name	Type	Description	Default
`x`	`NDArray[complex128]`	Samples to write. Must be 1-D and C-contiguous.	required

Returns:

Type	Description
`bool`	`True` if all samples were written; `False` if the ring was full and the call was refused (`x` untouched).

Examples:

>>> from doppler.buffer import F64Buffer
>>> import numpy as np
>>> buf = F64Buffer(512)
>>> buf.write(np.array([1+2j, 3+4j], dtype=np.complex128))
True
>>> buf2 = F64Buffer(512)
>>> buf2.write(np.zeros(512, dtype=np.complex128))
True
>>> buf2.write(np.zeros(1, dtype=np.complex128))
False

write_some ¶

write_some(x: NDArray[complex128]) -> int

Write as much of x as fits and say how much that was.

The partial-write twin of :meth:write. Where :meth:write refuses a block that does not fit whole, this takes the leading samples that do and returns their count -- 0 when the ring is full. It never refuses, so it never touches :attr:dropped. It is the only way to feed a chunk larger than the ring: loop, advancing by the return value, draining in between.

Parameters:

Name	Type	Description	Default
`x`	`NDArray[complex128]`	Samples to write. Must be 1-D and C-contiguous.	required

Returns:

Type	Description
`int`	Samples accepted, `0 <= k <= len(x)`. The caller still owns `x[k:]`.

Examples:

A chunk three times the size of the ring, fed by looping:

>>> from doppler.buffer import F64Buffer
>>> import numpy as np
>>> buf = F64Buffer(1024)
>>> cap = buf.capacity
>>> chunk = np.ones(3 * cap, dtype=np.complex128)
>>> fed = 0
>>> while fed < len(chunk):
...     fed += buf.write_some(chunk[fed:])
...     _ = buf.peek(buf.available); buf.consume()
>>> fed == 3 * cap, buf.dropped
(True, 0)

wait ¶

wait(n: int) -> NDArray[np.complex128]

Block until n samples are available; return zero-copy view.

Spins with the GIL released until the producer has written at least n samples. Returns a zero-copy 1-D complex128 view directly into the ring buffer. Caller must call :meth:consume before the next wait.

Parameters:

Name	Type	Description	Default
`n`	`int`	Number of complex samples to wait for.	required

Returns:

Type	Description
`NDArray[complex128]`	Zero-copy view into the ring buffer.

Raises:

Type	Description
`EOFError`	The producer called :meth:`close` and fewer than `n` samples remain. The tail is drained and no more is coming, so the wait ends rather than blocking forever.
`KeyboardInterrupt`	Somebody asked this process to stop, through a :class:`doppler.interrupt.Interrupt` guard -- from any module: the flag is process-wide. Without a guard the spin checks for no signals at all.

Examples:

>>> from doppler.buffer import F64Buffer
>>> import numpy as np
>>> buf = F64Buffer(512)
>>> buf.write(np.array([1+2j, 3+4j], dtype=np.complex128))
True
>>> view = buf.wait(2)
>>> view.dtype
dtype('complex128')
>>> view.shape
(2,)
>>> view.tolist()
[(1+2j), (3+4j)]
>>> buf.consume()

peek ¶

peek(n: int) -> NDArray[np.complex128] | None

:meth:wait that never blocks: a view, or None for not yet.

The single-threaded consumer's read. :meth:wait spins until a producer on another thread delivers, so a caller that is its own producer would deadlock in it; peek answers at once instead. When n samples are buffered it returns the same zero-copy, always-contiguous view :meth:wait would (1-D complex128); otherwise it returns None.

None means not yet and nothing else. The two conditions no amount of waiting can cure are raised, exactly as :meth:wait raises them, so a poll loop cannot mistake either for a slow producer.

Peeking does not consume. Follow it with :meth:consume; a consume(k) with k < n advances by a hop smaller than the frame, which is how overlapped frames are read.

Parameters:

Name	Type	Description	Default
`n`	`int`	Number of samples wanted. Must be positive and not larger than :attr:`capacity`.	required

Returns:

Type	Description
`NDArray[complex128] \| None`	Zero-copy view of the next `n` samples, or `None` when fewer than `n` have been written so far. None when the C call lends nothing (returns NULL) and its status has no exception row.

Raises:

Type	Description
`EOFError`	The ring is closed and fewer than `n` samples remain: the rest is never coming.
`ValueError`	`n` exceeds :attr:`capacity` (or is not positive), so no producer could ever satisfy it.

Examples:

>>> from doppler.buffer import F64Buffer
>>> import numpy as np
>>> buf = F64Buffer(1024)
>>> buf.peek(4) is None
True
>>> buf.write_some(np.ones(8, dtype=np.complex128))
8
>>> buf.peek(4).shape
(4,)
>>> buf.consume(2)
>>> buf.available
6
>>> buf.close()
>>> buf.peek(8)
Traceback (most recent call last):
    ...
EOFError: end of stream: the producer closed the ring

consume ¶

consume(n: int = ...) -> None

Release n samples back to the producer.

Advances the consumer tail pointer by n, making that space available for the producer to overwrite, and ends the loan: the view a :meth:wait or :meth:peek lent must not be used afterwards. If n is omitted it is the count of that outstanding view, so the number is written once. n smaller than the view is how overlapped frames are read: release a hop, keep the rest.

Parameters:

Name	Type	Description	Default
`n`	`int`	Number of samples to release. Defaults to the count of the outstanding :meth:`wait` / :meth:`peek` view.	`...`

Raises:

Type	Description
`ValueError`	`n` exceeds :attr:`available`. Nothing is released: past that point the ring's counts would stop describing it.
`RuntimeError`	`n` was omitted and nothing is outstanding -- no view was lent since the last release, so there is no count to default to.
`ValueError`	If the C call returns a non-zero status. The exception message is `consume(n): n exceeds the samples available; nothing was released`, with the return code appended (gh-869).

Examples:

>>> from doppler.buffer import F64Buffer
>>> import numpy as np
>>> buf = F64Buffer(512)
>>> buf.write(np.ones(4, dtype=np.complex128))
True
>>> _ = buf.wait(4)
>>> buf.consume()

close ¶

close() -> None

Say that no more data is coming.

The producer's half of end of stream. Until this exists a consumer cannot tell a slow producer from a finished one -- both look like an empty ring -- so :meth:wait had nothing to do but spin. Call it once, after the last write.

Release ordering: every sample written before this is visible to a consumer that observes the flag. Closing does not discard what was already written; :meth:wait keeps returning batches until the ring is drained, and only then raises EOFError.

See docs/design/io-termination.md for the one termination contract shared with the network and disk transports.

Examples:

>>> import numpy as np
>>> from doppler.buffer import F64Buffer
>>> buf = F64Buffer(1024)
>>> buf.close()
>>> buf.closed
True
>>> buf.wait(4)
Traceback (most recent call last):
    ...
EOFError: end of stream: the producer closed the ring

reset ¶

reset() -> None

Empty the ring and reopen it.

Discards everything buffered, and clears :attr:closed so the same ring can carry a second stream -- without it, reuse after :meth:close means destroying and re-mapping. :attr:dropped is a lifetime count and is kept.

Not safe against a concurrent producer or consumer: it moves both ends of the ring. Call it only when both sides are idle.

Examples:

>>> from doppler.buffer import F64Buffer
>>> import numpy as np
>>> buf = F64Buffer(1024)
>>> buf.write_some(np.ones(8, dtype=np.complex128))
8
>>> buf.close()
>>> buf.reset()
>>> buf.available, buf.closed
(0, False)

destroy ¶

destroy() -> None

Close the object; the memory goes with its last user.

Ordinarily unnecessary: the resources are freed when the object is garbage-collected. Call this, or use the object as a context manager, to end its use at a definite point.

Views this object handed out keep the memory they point into, as does a call still running on another thread: it is freed when the last of them, and this object, are gone, not here (gh-2187). Idempotent: calling it again does nothing. Every other method raises RuntimeError once it has run.

enter ¶

__enter__() -> F64Buffer

Enter a context manager, returning this object.

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

Returns:

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

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

I16Buffer ¶

Lock-free SPSC ring buffer for interleaved int16 IQ pairs.

Parameters:

Name	Type	Description	Default
`capacity`	`int`	How many samples the ring holds: any size from 1 up, and :attr:`capacity` is exactly this number on every machine. What is rounded is the MAPPING behind it -- up to a power of two, because indexing is a mask, and up to a whole page -- so a capacity that is not a power of two costs some address space (under 2x) and nothing per call.	required

Raises:

Type	Description
`ValueError`	If construction fails. The exception message is `capacity must be at least 1, and small enough to map`.

Examples:

>>> from doppler.buffer import I16Buffer
>>> import numpy as np
>>> buf = I16Buffer(1024)
>>> buf.capacity
1024
>>> IQ16 = np.dtype([("i", "<i2"), ("q", "<i2")])
>>> adc = np.array([10, 20, 30, 40], dtype=np.int16)   # I, Q, I, Q
>>> buf.write(adc.view(IQ16))
True
>>> buf.wait(2)["q"].tolist()
[20, 40]

capacity `property` ¶

capacity: int

Buffer capacity in IQ sample pairs.

available `property` ¶

available: int

Samples written but not yet consumed.

space `property` ¶

space: int

Free room in samples: the largest :meth:write sure to fit.

dropped `property` ¶

dropped: int

Cumulative IQ sample pairs in REFUSED writes -- not pairs lost.

closed `property` ¶

closed: bool

True once the producer has called :meth:close.

write ¶

write(x: NDArray[Any]) -> bool

Write IQ samples into the buffer without blocking.

Copies the record array into the ring in a single memcpy. With fewer than len(x) free slots the call is refused entirely -- nothing copied, x untouched -- and :attr:dropped grows by len(x), which is not a loss count: you still hold every sample.

A bare int16 array is refused with TypeError, flat or (n, 2): it is not an array of samples. flat.view(IQ16) makes it one, with no copy.

Parameters:

Name	Type	Description	Default
`x`	`NDArray[Any]`	IQ samples to write: 1-D, C-contiguous, dtype `[("i", "<i2"), ("q", "<i2")]`.	required

Returns:

Type	Description
`bool`	`True` if all samples were written; `False` if the ring had no room and the call was refused (`x` untouched).

Examples:

>>> from doppler.buffer import I16Buffer
>>> import numpy as np
>>> buf = I16Buffer(1024)
>>> IQ16 = np.dtype([("i", "<i2"), ("q", "<i2")])
>>> buf.write(np.array([10, 20, 30, 40], dtype=np.int16).view(IQ16))
True
>>> buf2 = I16Buffer(1024)
>>> buf2.write(np.zeros(1024, dtype=IQ16))
True
>>> buf2.write(np.zeros(1, dtype=IQ16))
False

write_some ¶

write_some(x: NDArray[Any]) -> int

Write as much of x as fits and say how much that was.

The partial-write twin of :meth:write. Where :meth:write refuses a block that does not fit whole, this takes the leading samples that do and returns their count -- 0 when the ring is full. It never refuses, so it never touches :attr:dropped. It is the only way to feed a chunk larger than the ring: loop, advancing by the return value, draining in between.

Parameters:

Name	Type	Description	Default
`x`	`NDArray[Any]`	Samples to write: 1-D, C-contiguous, dtype `[("i", "<i2"), ("q", "<i2")]`.	required

Returns:

Type	Description
`int`	Samples accepted, `0 <= k <= len(x)`. The caller still owns `x[k:]`.

Examples:

A chunk three times the size of the ring, fed by looping:

>>> from doppler.buffer import I16Buffer
>>> import numpy as np
>>> IQ16 = np.dtype([("i", "<i2"), ("q", "<i2")])
>>> buf = I16Buffer(1024)
>>> cap = buf.capacity
>>> chunk = np.ones(3 * cap, dtype=IQ16)
>>> fed = 0
>>> while fed < len(chunk):
...     fed += buf.write_some(chunk[fed:])
...     _ = buf.peek(buf.available); buf.consume()
>>> fed == 3 * cap, buf.dropped
(True, 0)

wait ¶

wait(n: int) -> NDArray[Any]

Block until n samples are available, then lend a zero-copy view.

Spins with the GIL released until the producer has written at least n samples. Returns a 1-D record array directly into the double-mapped ring: view["i"] is the I channel, view["q"] the Q channel, each a strided int16 view with no copy. Caller must call :meth:consume before the next wait.

Parameters:

Name	Type	Description	Default
`n`	`int`	Number of IQ sample pairs to wait for.	required

Returns:

Type	Description
`NDArray[Any]`	Zero-copy view of the next `n` samples, one record each.

Raises:

Type	Description
`EOFError`	The producer called :meth:`close` and fewer than `n` samples remain. The tail is drained and no more is coming, so the wait ends rather than blocking forever.
`KeyboardInterrupt`	Somebody asked this process to stop, through a :class:`doppler.interrupt.Interrupt` guard -- from any module: the flag is process-wide. Without a guard the spin checks for no signals at all.

Examples:

>>> from doppler.buffer import I16Buffer
>>> import numpy as np
>>> buf = I16Buffer(1024)
>>> IQ16 = np.dtype([("i", "<i2"), ("q", "<i2")])
>>> buf.write(np.array([10, 20, 30, 40], dtype=np.int16).view(IQ16))
True
>>> view = buf.wait(2)
>>> view.dtype
dtype([('i', '<i2'), ('q', '<i2')])
>>> view.shape
(2,)
>>> view.tolist()
[(10, 20), (30, 40)]
>>> buf.consume()

peek ¶

peek(n: int) -> NDArray[Any] | None

:meth:wait that never blocks: a view, or None for not yet.

The single-threaded consumer's read. :meth:wait spins until a producer on another thread delivers, so a caller that is its own producer would deadlock in it; peek answers at once instead. When n samples are buffered it returns the same zero-copy, always-contiguous view :meth:wait would (1-D, one (i, q) record per sample); otherwise it returns None.

None means not yet and nothing else. The two conditions no amount of waiting can cure are raised, exactly as :meth:wait raises them, so a poll loop cannot mistake either for a slow producer.

Peeking does not consume. Follow it with :meth:consume; a consume(k) with k < n advances by a hop smaller than the frame, which is how overlapped frames are read.

Parameters:

Name	Type	Description	Default
`n`	`int`	Number of samples wanted. Must be positive and not larger than :attr:`capacity`.	required

Returns:

Type	Description
`NDArray[Any] \| None`	Zero-copy view of the next `n` samples, or `None` when fewer than `n` have been written so far. None when the C call lends nothing (returns NULL) and its status has no exception row.

Raises:

Type	Description
`EOFError`	The ring is closed and fewer than `n` samples remain: the rest is never coming.
`ValueError`	`n` exceeds :attr:`capacity` (or is not positive), so no producer could ever satisfy it.

Examples:

>>> from doppler.buffer import I16Buffer
>>> import numpy as np
>>> IQ16 = np.dtype([("i", "<i2"), ("q", "<i2")])
>>> buf = I16Buffer(1024)
>>> buf.peek(4) is None
True
>>> buf.write_some(np.ones(8, dtype=IQ16))
8
>>> buf.peek(4).shape
(4,)
>>> buf.consume(2)
>>> buf.available
6
>>> buf.close()
>>> buf.peek(8)
Traceback (most recent call last):
    ...
EOFError: end of stream: the producer closed the ring

consume ¶

consume(n: int = ...) -> None

Release n samples back to the producer.

Advances the consumer tail pointer by n, making that space available for the producer to overwrite, and ends the loan: the view a :meth:wait or :meth:peek lent must not be used afterwards. If n is omitted it is the count of that outstanding view, so the number is written once. n smaller than the view is how overlapped frames are read: release a hop, keep the rest.

Parameters:

Name	Type	Description	Default
`n`	`int`	Number of samples to release. Defaults to the count of the outstanding :meth:`wait` / :meth:`peek` view.	`...`

Raises:

Type	Description
`ValueError`	`n` exceeds :attr:`available`. Nothing is released: past that point the ring's counts would stop describing it.
`RuntimeError`	`n` was omitted and nothing is outstanding -- no view was lent since the last release, so there is no count to default to.
`ValueError`	If the C call returns a non-zero status. The exception message is `consume(n): n exceeds the samples available; nothing was released`, with the return code appended (gh-869).

Examples:

>>> from doppler.buffer import I16Buffer
>>> import numpy as np
>>> IQ16 = np.dtype([("i", "<i2"), ("q", "<i2")])
>>> buf = I16Buffer(1024)
>>> buf.write(np.array([1, 2, 3, 4], dtype=np.int16).view(IQ16))
True
>>> _ = buf.wait(2)
>>> buf.consume()

close ¶

close() -> None

Say that no more data is coming.

The producer's half of end of stream. Until this exists a consumer cannot tell a slow producer from a finished one -- both look like an empty ring -- so :meth:wait had nothing to do but spin. Call it once, after the last write.

Release ordering: every sample written before this is visible to a consumer that observes the flag. Closing does not discard what was already written; :meth:wait keeps returning batches until the ring is drained, and only then raises EOFError.

See docs/design/io-termination.md for the one termination contract shared with the network and disk transports.

Examples:

>>> import numpy as np
>>> from doppler.buffer import I16Buffer
>>> buf = I16Buffer(1024)
>>> buf.close()
>>> buf.closed
True
>>> buf.wait(4)
Traceback (most recent call last):
    ...
EOFError: end of stream: the producer closed the ring

reset ¶

reset() -> None

Empty the ring and reopen it.

Discards everything buffered, and clears :attr:closed so the same ring can carry a second stream -- without it, reuse after :meth:close means destroying and re-mapping. :attr:dropped is a lifetime count and is kept.

Not safe against a concurrent producer or consumer: it moves both ends of the ring. Call it only when both sides are idle.

Examples:

>>> from doppler.buffer import I16Buffer
>>> import numpy as np
>>> IQ16 = np.dtype([("i", "<i2"), ("q", "<i2")])
>>> buf = I16Buffer(1024)
>>> buf.write_some(np.ones(8, dtype=IQ16))
8
>>> buf.close()
>>> buf.reset()
>>> buf.available, buf.closed
(0, False)

destroy ¶

destroy() -> None

Close the object; the memory goes with its last user.

Ordinarily unnecessary: the resources are freed when the object is garbage-collected. Call this, or use the object as a context manager, to end its use at a definite point.

Views this object handed out keep the memory they point into, as does a call still running on another thread: it is freed when the last of them, and this object, are gone, not here (gh-2187). Idempotent: calling it again does nothing. Every other method raises RuntimeError once it has run.

enter ¶

__enter__() -> I16Buffer

Enter a context manager, returning this object.

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

Returns:

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

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

Guides — Real-Time Pacing & Timestamping Design — The ring buffer — one contiguous view of a stream, Spectrum Analyzer

Python Ring Buffer API¶

Buffer types¶

Two surfaces, one ring¶

Examples¶

Producer / consumer (threaded)¶

One thread: any block in, exact frames out¶

One thread: a chunk larger than the ring, overlapped frames out¶

End of stream, then the same ring again¶

I16Buffer — raw ADC samples¶

A full ring¶

F32Buffer ¶

capacity property ¶

available property ¶

space property ¶

dropped property ¶

closed property ¶

write ¶

write_some ¶

wait ¶

peek ¶

consume ¶

close ¶

reset ¶

destroy ¶

__enter__ ¶

__exit__ ¶

F64Buffer ¶

capacity property ¶

available property ¶

space property ¶

dropped property ¶

closed property ¶

write ¶

write_some ¶

wait ¶

peek ¶

consume ¶

close ¶

reset ¶

destroy ¶

__enter__ ¶

__exit__ ¶

I16Buffer ¶

capacity property ¶

available property ¶

space property ¶

dropped property ¶

closed property ¶

write ¶

write_some ¶

wait ¶

peek ¶

consume ¶

close ¶

reset ¶

destroy ¶

__enter__ ¶

__exit__ ¶

Related pages¶

capacity `property` ¶

available `property` ¶

space `property` ¶

dropped `property` ¶

closed `property` ¶

enter ¶

exit ¶

capacity `property` ¶

available `property` ¶

space `property` ¶

dropped `property` ¶

closed `property` ¶

enter ¶

exit ¶

capacity `property` ¶

available `property` ¶

space `property` ¶

dropped `property` ¶

closed `property` ¶

enter ¶

exit ¶