Python Streaming API¶

doppler.stream moves sample blocks between processes (or hosts) over NATS, in three subject-based patterns that share one wire format — a dp_header_t (sample rate, centre frequency, sample type, an auto timestamp_ns, and a per-socket sequence counter) followed by the raw sample bytes. Because the header is shared, a C transmitter and a Python receiver interoperate freely.

Every pattern needs a nats-server reachable at the endpoint's host:port — plain nats-server is enough for PUB/SUB and REQ/REP; PUSH/PULL rides the JetStream work-queue tier, so start it with nats-server -js. Endpoints are nats://host:port/subject.

Source: src/doppler/stream/__init__.py

For the C side and a runnable end-to-end walk-through, see the Streaming examples.

Socket patterns¶

Pattern	Sender	Receiver	NATS tier	Use
PUB / SUB	`Publisher`	`Subscriber`	Core NATS	fan-out broadcast; subscribers may drop
PUSH/PULL	`Push`	`Pull`	JetStream work-queue	load-balanced pipeline; durable, acked
REQ / REP	`Requester`	`Replier`	NATS request/reply	request/response (lock-step)

Every socket is a context manager and releases the GIL while blocked waiting on NATS, so receive loops thread cleanly. The sender's sample_type fixes the NumPy dtype on both ends:

Constant	dtype	layout
`CF32`	`numpy.complex64`	complex I/Q
`CF64`	`numpy.complex128`	complex I/Q
`CF128`	`numpy.clongdouble`	extended-precision complex
`CI8`	`numpy.int8`	interleaved I/Q, length `2n`
`CI16`	`numpy.int16`	interleaved I/Q, length `2n`
`CI32`	`numpy.int32`	interleaved I/Q, length `2n`
`TLM16`	structured rows	telemetry records, not I/Q (Telemetry)

recv() returns (samples, header) where header is a dict of the decoded dp_header_t fields (sample_rate, center_freq, sample_type, timestamp_ns, sequence).

PUB / SUB — broadcast¶

The publisher fans out to every subscriber on the subject; a slow subscriber drops frames rather than back-pressuring the sender. Core NATS has no backlog, so a subscriber must already be listening before the publish — start the Subscriber first (or add a brief warm-up sleep).

import numpy as np
from doppler.stream import Publisher, Subscriber, CF64

# transmitter
with Publisher("nats://127.0.0.1:4222/iq", CF64) as pub:
    iq = np.exp(2j * np.pi * 1e3 * np.arange(1000) / 1e6)   # complex128
    pub.send(iq, sample_rate=1e6, center_freq=2.4e9)

# receiver — in another process, started before the publish above
with Subscriber("nats://127.0.0.1:4222/iq") as sub:
    samples, header = sub.recv(timeout_ms=1000)
    print(header["sample_rate"], header["sequence"], len(samples))

PUSH / PULL — pipeline¶

PUSH publishes onto a durable JetStream work-queue subject; JetStream load-balances each frame to exactly one connected PULL worker, giving a back-pressured, at-least-once work queue. Call Pull.ack() once a frame has been fully processed. Requires a JetStream-enabled broker (nats-server -js).

import numpy as np
from doppler.stream import Push, Pull, CF64

with Push("nats://127.0.0.1:4222/work", CF64) as push:
    push.send(np.zeros(4096, dtype=np.complex128), sample_rate=1e6)

with Pull("nats://127.0.0.1:4222/work") as pull:
    samples, header = pull.recv()       # blocks until a frame arrives
    pull.ack(samples)

REQ / REP — request/response¶

A Requester sends a sample block and blocks for the Replier's response; the two alternate strictly, over NATS's native request/reply.

import numpy as np
from doppler.stream import Requester, Replier, CF64

# server: receive a request, send a reply
with Replier("nats://127.0.0.1:4222/ctrl", CF64) as rep:
    req, header = rep.recv()
    rep.send(req * 2)                   # echo-and-scale, say

# client: send, then receive the reply
with Requester("nats://127.0.0.1:4222/ctrl", CF64) as req:
    req.send(np.ones(256, dtype=np.complex128), sample_rate=1e6)
    resp, header = req.recv()

Timestamps¶

get_timestamp_ns() returns the same CLOCK_REALTIME nanosecond stamp the senders embed in each header, so a receiver can compute end-to-end latency against header["timestamp_ns"].

PUB / SUB¶

Publisher ¶

NATS core PUB — one-to-many broadcast of signal frames.

Wraps dp_pub_t. Each :meth:send call stages one [dp_header_t][raw data] buffer and publishes it to the endpoint's iq.<subject>.<type> NATS subject. Multiple :class:Subscriber sockets can subscribe to the same subject; each subscriber receives every frame. NATS core has no persistence or back-pressure: a slow subscriber simply misses frames published while it isn't reading.

The endpoint identifies a subject on a NATS broker; both Publisher and Subscriber connect to the same nats-server (no bind/connect distinction — the broker mediates fan-out).

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	NATS endpoint, e.g. `"nats://127.0.0.1:4222/iq"` (`host:port/subject`; the subject defaults to `"default"` if omitted). Requires a running `nats-server`.	required
`sample_type`	`int`	Wire encoding. One of :data:`CI32`, :data:`CF64` (default), :data:`CF128`. Raises :exc:`ValueError` for any other value.	`...`

Raises:

Type	Description
`ValueError`	If `sample_type` is not one of the three supported values.
`RuntimeError`	If `nats-server` isn't reachable at `endpoint`.

Examples:

Construct, use as a context manager, and verify the type:

>>> from doppler.stream import Publisher, CF64
>>> pub = Publisher("nats://127.0.0.1:4222/t19100", CF64)
>>> type(pub).__name__
'Publisher'
>>> pub.close()

Context-manager form (preferred — ensures close on exception):

>>> with Publisher("nats://127.0.0.1:4222/t19101", CF64) as pub:
...     type(pub).__name__
'Publisher'

Full send round-trip with a :class:Subscriber (requires a live nats-server and a brief warm-up sleep so the subscription is established before the first publish):

>>> import numpy as np, time
>>> from doppler.stream import Subscriber
>>> pub = Publisher("nats://127.0.0.1:4222/iq", CF64)
>>> sub = Subscriber("nats://127.0.0.1:4222/iq")
>>> time.sleep(0.1)
>>> pub.send(np.array([1+2j, 3+4j],
...          dtype=np.complex128),
...          sample_rate=int(1e6),
...          center_freq=int(2.4e9))
>>> samples, hdr = sub.recv(timeout_ms=2000)
>>> pub.close(); sub.close()

init ¶

__init__(endpoint: str, sample_type: int = ...) -> None

Create a Publisher and connect to endpoint's NATS broker.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	NATS endpoint, e.g. `"nats://127.0.0.1:4222/iq"`.	required
`sample_type`	`int`	Wire encoding: :data:`CI32`, :data:`CF64` (default), or :data:`CF128`.	`...`

enter ¶

__enter__() -> Publisher

Return self for use as a context manager.

Examples:

>>> from doppler.stream import Publisher, CF64
>>> with Publisher("nats://127.0.0.1:4222/t19102", CF64) as pub:
...     type(pub).__name__
'Publisher'

exit ¶

__exit__(*args: object) -> None

Call :meth:close on context-manager exit.

send ¶

send(
    samples: NDArray[Any],
    sample_rate: float = 0,
    center_freq: float = 0,
    timestamp_ns: int | None = None,
) -> None

Broadcast one block of samples to all connected Subscribers.

Constructs a dp_header_t with the supplied metadata (plus a per-socket monotonically increasing sequence number), stages one [header][raw sample bytes] buffer (NATS has no scatter/ gather send, so header and data are copied into one contiguous buffer first), and publishes it. The call releases the GIL while blocked in the underlying NATS client, so other Python threads can run concurrently.

Parameters:

Name	Type	Description	Default
`samples`	`ndarray`	C-contiguous array whose dtype must match the socket's `sample_type`: `numpy.complex128` for :data:`CF64`, `numpy.clongdouble` for :data:`CF128`, `numpy.int32` for :data:`CI32` (interleaved I/Q, length `2*n_samples`).	required
`sample_rate`	`float`	Samples per second written into the header (default 0).	`0`
`center_freq`	`float`	Centre frequency in Hz written into the header (default 0).	`0`
`timestamp_ns`	`int`	UNIX nanoseconds to stamp on `timestamp_ns` instead of a fresh `CLOCK_REALTIME` read -- propagate an upstream message's own origin timestamp (or one derived from :class:`~doppler.wfm.SampleClock`) through this hop instead of overwriting it with "now" (default `None`: auto-stamp).	`None`

Raises:

Type	Description
`TypeError`	If `samples.dtype` does not match the socket's `sample_type`.
`ValueError`	If `samples` is not C-contiguous.
`RuntimeError`	If the publish fails (e.g. the broker connection dropped).

Examples:

>>> from doppler.stream import Publisher, CF64
>>> import numpy as np, time
>>> pub = Publisher("nats://127.0.0.1:4222/iq", CF64)
>>> pub.send(np.ones(4, dtype=np.complex128),
...          sample_rate=int(48000),
...          center_freq=int(433e6))
>>> pub.close()

close ¶

close() -> None

Destroy the underlying NATS handle and release all resources.

Calls dp_pub_destroy(). Safe to call multiple times — subsequent calls are no-ops. After close() the object must not be used for sending.

Examples:

>>> from doppler.stream import Publisher, CF64
>>> pub = Publisher("nats://127.0.0.1:4222/t19103", CF64)
>>> pub.close()
>>> pub.close()  # idempotent — no error

Subscriber ¶

NATS core SUB — receives signal frames from one or more Publishers.

Wraps dp_sub_t. Subscribes to the endpoint's iq.<subject>.> NATS subject, so it receives every frame any :class:Publisher bound to that subject sends (fan-out: every Subscriber on the subject gets every frame). There is no "connect to exactly one Publisher" concept — the broker mediates delivery by subject, not by socket.

For load-balanced consumption (each frame delivered to exactly one of several workers, with durable redelivery) use :class:Pull instead, which is backed by NATS JetStream rather than NATS core.

The recv path is zero-copy: the returned NumPy array's memory is owned by an internal dp_msg_t handle that points into the received NATS message; the buffer is freed only when the array (and all views of it) are garbage-collected.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	NATS endpoint, e.g. `"nats://127.0.0.1:4222/iq"` (`host:port/subject`; the subject defaults to `"default"` if omitted). Requires a running `nats-server`.	required

Raises:

Type	Description
`RuntimeError`	If `nats-server` isn't reachable at `endpoint`.

Examples:

>>> from doppler.stream import Subscriber
>>> sub = Subscriber("nats://127.0.0.1:4222/t19104")
>>> type(sub).__name__
'Subscriber'
>>> sub.close()

Context-manager form:

>>> with Subscriber("nats://127.0.0.1:4222/t19105") as sub:
...     type(sub).__name__
'Subscriber'

Receive one frame (requires a live :class:Publisher):

>>> sub.recv(timeout_ms=500)
(array([1.+2.j, 3.+4.j]), {'sample_rate': 1000000.0, ...})

init ¶

__init__(endpoint: str) -> None

Create a Subscriber socket and connect to endpoint.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	NATS endpoint, e.g. `"nats://127.0.0.1:4222/iq"`.	required

enter ¶

__enter__() -> Subscriber

Return self for use as a context manager.

Examples:

>>> from doppler.stream import Subscriber
>>> with Subscriber("nats://127.0.0.1:4222/t19106") as sub:
...     type(sub).__name__
'Subscriber'

exit ¶

__exit__(*args: object) -> None

Call :meth:close on context-manager exit.

recv ¶

recv(
    timeout_ms: int = -1,
) -> Tuple[NDArray[Any], Dict[str, Any]]

Receive one signal frame from the connected Publisher.

Blocks until a frame arrives or the optional timeout expires. The returned NumPy array is a zero-copy view into the received NATS message buffer; the buffer is freed when the array is garbage-collected.

Parameters:

Name	Type	Description	Default
`timeout_ms`	`int`	Milliseconds to wait before raising :exc:`TimeoutError`. `-1` (default) blocks indefinitely; `0` returns immediately if no frame is available.	`-1`

Returns:

Name	Type	Description
`samples`	`ndarray`	Decoded sample data. dtype is `numpy.complex128` (:data:`CF64`), `numpy.clongdouble` (:data:`CF128`), or `numpy.int32` flat interleaved I/Q (:data:`CI32`).
`header`	`dict`	Decoded `dp_header_t` fields: `sample_rate` : float Samples per second as reported by the sender. `center_freq` : float Centre frequency in Hz as reported by the sender. `sample_type` : int Wire sample type (one of :data:`CI32`, :data:`CF64`, :data:`CF128`). `timestamp_ns` : int Frame timestamp (`CLOCK_REALTIME` nanoseconds) set by the sender at the moment of the send call. `sequence` : int Monotonically increasing per-sender frame counter; gaps indicate dropped frames. `num_samples` : int Number of IQ samples in the frame (`len(samples)` for CF64/CF128; `len(samples)//2` for CI32). `protocol` : int Wire protocol (0 = SIGS, 1 = DIFI/VITA 49). `stream_id` : int Opaque stream identifier (0 for SIGS protocol).

Raises:

Type	Description
`TimeoutError`	If `timeout_ms` elapses before a frame arrives.
`RuntimeError`	If the underlying NATS recv fails for any other reason.

Examples:

>>> from doppler.stream import Subscriber
>>> with Subscriber("nats://127.0.0.1:4222/iq") as sub:
...     samples, hdr = sub.recv(timeout_ms=1000)
...     print(samples.dtype, hdr["sample_rate"])

close ¶

close() -> None

Destroy the underlying NATS handle and release all resources.

Calls dp_sub_destroy(). Safe to call multiple times.

Examples:

>>> from doppler.stream import Subscriber
>>> sub = Subscriber("nats://127.0.0.1:4222/t19107")
>>> sub.close()
>>> sub.close()  # idempotent — no error

PUSH / PULL¶

Push ¶

NATS JetStream work-queue producer — durable pipeline sender.

Wraps dp_push_t. Each :meth:send is a synchronous, server- acked JetStream publish: the frame is persisted (and replicated, on a clustered broker) before the call returns, so a producer crash never silently drops a frame. Each persisted frame is later delivered to exactly one :class:Pull worker (competing-consumers distribution), unlike :class:Publisher which fans out to every Subscriber.

Use the PUSH/PULL pattern when you have a pool of workers pulling from a shared, durable queue and want at-least-once delivery with redelivery on worker crash (see :meth:Pull.ack).

Both ends address the same NATS work-queue subject; there is no bind/connect distinction (the broker mediates delivery).

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	NATS endpoint, e.g. `"nats://127.0.0.1:4222/work"`.	required
`sample_type`	`int`	Wire encoding. One of :data:`CI32`, :data:`CF64` (default), :data:`CF128`. Raises :exc:`ValueError` for any other value.	`...`

Raises:

Type	Description
`ValueError`	If `sample_type` is not one of the three supported values.
`RuntimeError`	If `nats-server` isn't reachable at `endpoint`.

Examples:

>>> from doppler.stream import Push, CF64
>>> push = Push("nats://127.0.0.1:4222/t19108", CF64)
>>> type(push).__name__
'Push'
>>> push.close()

Context-manager form:

>>> with Push("nats://127.0.0.1:4222/t19109", CF64) as push:
...     type(push).__name__
'Push'

Round-trip with a :class:Pull worker (requires a live connection):

>>> import numpy as np, time
>>> from doppler.stream import Pull
>>> push = Push("nats://127.0.0.1:4222/work", CF64)
>>> pull = Pull("nats://127.0.0.1:4222/work")
>>> time.sleep(0.05)
>>> push.send(np.ones(4, dtype=np.complex128))
>>> samples, hdr = pull.recv(timeout_ms=2000)
>>> push.close(); pull.close()

init ¶

__init__(endpoint: str, sample_type: int = ...) -> None

Create a Push socket and bind to endpoint.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	NATS endpoint, e.g. `"nats://127.0.0.1:4222/work"`.	required
`sample_type`	`int`	Wire encoding: :data:`CI32`, :data:`CF64` (default), or :data:`CF128`.	`...`

enter ¶

__enter__() -> Push

Return self for use as a context manager.

Examples:

>>> from doppler.stream import Push, CF64
>>> with Push("nats://127.0.0.1:4222/t19110", CF64) as push:
...     type(push).__name__
'Push'

exit ¶

__exit__(*args: object) -> None

Call :meth:close on context-manager exit.

send ¶

send(
    samples: NDArray[Any],
    sample_rate: float = 0,
    center_freq: float = 0,
    timestamp_ns: int | None = None,
) -> None

Durably publish one block of samples to the work-queue.

Blocks until the broker acks that the frame is persisted (and replicated, on a clustered broker) — not until a Pull worker is ready; the frame waits in the queue for the next available worker. Releases the GIL while blocked on the broker round trip.

Parameters:

Name	Type	Description	Default
`samples`	`ndarray`	C-contiguous array whose dtype must match the socket's `sample_type`: `numpy.complex128` for :data:`CF64`, `numpy.clongdouble` for :data:`CF128`, `numpy.int32` for :data:`CI32` (interleaved I/Q, length `2*n`).	required
`sample_rate`	`float`	Samples per second written into the header (default 0).	`0`
`center_freq`	`float`	Centre frequency in Hz written into the header (default 0).	`0`
`timestamp_ns`	`int`	UNIX nanoseconds to stamp on `timestamp_ns` instead of a fresh `CLOCK_REALTIME` read -- propagate an upstream message's own origin timestamp through this hop instead of overwriting it with "now" (default `None`: auto-stamp).	`None`

Raises:

Type	Description
`TypeError`	If `samples.dtype` does not match the socket's `sample_type`.
`ValueError`	If `samples` is not C-contiguous.
`RuntimeError`	If the publish fails (e.g. the broker connection dropped).

Examples:

>>> from doppler.stream import Push, Pull, CF64
>>> import numpy as np, time
>>> push = Push("nats://127.0.0.1:4222/work", CF64)
>>> pull = Pull("nats://127.0.0.1:4222/work")
>>> time.sleep(0.05)
>>> push.send(np.array([1+2j, 3+4j],
...           dtype=np.complex128),
...           sample_rate=int(48000))
>>> samples, hdr = pull.recv(timeout_ms=2000)
>>> push.close(); pull.close()

close ¶

close() -> None

Destroy the underlying NATS handle and release all resources.

Calls dp_push_destroy(). Safe to call multiple times.

Examples:

>>> from doppler.stream import Push, CF64
>>> push = Push("nats://127.0.0.1:4222/t19111", CF64)
>>> push.close()
>>> push.close()  # idempotent — no error

Pull ¶

NATS JetStream work-queue consumer — durable pipeline receiver.

Wraps dp_pull_t. Consumes frames persisted by one or more :class:Push producers. Multiple Pull workers can share the same durable consumer group; each frame is delivered to exactly one worker. Call :meth:ack once a frame is fully processed — an un-acked frame is redelivered if the worker dies first.

The recv path is zero-copy: see :class:Subscriber for the buffer lifetime semantics.

Both ends address the same NATS work-queue subject; there is no bind/connect distinction (the broker mediates delivery).

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	NATS endpoint, e.g. `"nats://127.0.0.1:4222/work"`. Requires a running `nats-server -js` (JetStream enabled).	required

Raises:

Type	Description
`RuntimeError`	If `nats-server` isn't reachable at `endpoint`.

Examples:

>>> from doppler.stream import Push, Pull, CF64
>>> _ = Push("nats://127.0.0.1:4222/t19112", CF64)  # provisions the queue
>>> pull = Pull("nats://127.0.0.1:4222/t19112")
>>> type(pull).__name__
'Pull'
>>> pull.close()

Context-manager form:

>>> _ = Push("nats://127.0.0.1:4222/t19113", CF64)  # provisions the queue
>>> with Pull("nats://127.0.0.1:4222/t19113") as pull:
...     type(pull).__name__
'Pull'

Receive one frame (requires a live :class:Push):

>>> pull.recv(timeout_ms=500)
(array([1.+2.j, ...]), {'sample_rate': ..., ...})

init ¶

__init__(endpoint: str) -> None

Create a Pull socket and connect to endpoint.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	NATS endpoint, e.g. `"nats://127.0.0.1:4222/work"`.	required

enter ¶

__enter__() -> Pull

Return self for use as a context manager.

Examples:

>>> from doppler.stream import Push, Pull, CF64
>>> _ = Push("nats://127.0.0.1:4222/t19114", CF64)  # provisions queue
>>> with Pull("nats://127.0.0.1:4222/t19114") as pull:
...     type(pull).__name__
'Pull'

exit ¶

__exit__(*args: object) -> None

Call :meth:close on context-manager exit.

recv ¶

recv(
    timeout_ms: int = -1,
) -> Tuple[NDArray[Any], Dict[str, Any]]

Receive one signal frame from the connected Push socket.

Blocks until a frame arrives or the optional timeout expires. The returned NumPy array is a zero-copy view into the received NATS message buffer; the buffer is freed when the array is garbage-collected.

Parameters:

Name	Type	Description	Default
`timeout_ms`	`int`	Milliseconds to wait before raising :exc:`TimeoutError`. `-1` (default) blocks indefinitely; `0` returns immediately if no frame is available.	`-1`

Returns:

Name	Type	Description
`samples`	`ndarray`	Decoded sample data. dtype is `numpy.complex128` (:data:`CF64`), `numpy.clongdouble` (:data:`CF128`), or `numpy.int32` flat interleaved I/Q (:data:`CI32`).
`header`	`dict`	Decoded `dp_header_t` fields — see :meth:`Subscriber.recv` for the full key list.

Raises:

Type	Description
`TimeoutError`	If `timeout_ms` elapses before a frame arrives.
`RuntimeError`	If the underlying NATS recv fails for any other reason.

Examples:

>>> from doppler.stream import Pull
>>> with Pull("nats://127.0.0.1:4222/work") as pull:
...     samples, hdr = pull.recv(timeout_ms=1000)
...     print(samples.dtype, hdr["num_samples"])

ack ¶

ack(samples: NDArray[Any]) -> None

Acknowledge a frame consumed from the JetStream work-queue.

Delivery is at-least-once: a frame stays pending until acked and is redelivered to another worker if this one dies first. Pass the array returned by :meth:recv once it has been fully processed, then drop the array.

Parameters:

Name	Type	Description	Default
`samples`	`ndarray`	The samples array returned by :meth:`recv` (its buffer still alive — do not ack after the array is garbage-collected).	required

Raises:

Type	Description
`ValueError`	If `samples` is not an un-freed :meth:`recv` result.
`RuntimeError`	If the acknowledgement fails.

Examples:

>>> from doppler.stream import Pull
>>> pull = Pull("nats://localhost:4222/work")
>>> samples, hdr = pull.recv(timeout_ms=1000)
>>> pull.ack(samples)

close ¶

close() -> None

Destroy the underlying NATS handle and release all resources.

Calls dp_pull_destroy(). Safe to call multiple times.

Examples:

>>> from doppler.stream import Push, Pull, CF64
>>> _ = Push("nats://127.0.0.1:4222/t19115", CF64)  # provisions queue
>>> pull = Pull("nats://127.0.0.1:4222/t19115")
>>> pull.close()
>>> pull.close()  # idempotent — no error

REQ / REP¶

Requester ¶

NATS request/reply — sends a request frame, then waits for a reply.

Wraps dp_req_t. Built on a NATS request: :meth:send publishes to the endpoint's subject with a reply-to address (a dedicated inbox created for this Requester), and :meth:recv waits on that same inbox for the :class:Replier's answer. Use this pattern strictly alternating (send, then recv, then send again) — unlike ZMQ's REQ socket there is no enforced state machine, but calling recv before a matching send simply blocks/times out on an empty inbox rather than raising immediately.

Complements :class:Replier. Use this pattern for control-plane messages (tuning commands, metadata queries) or synchronous signal-frame RPC where one peer processes each frame and returns a result.

Both ends address the same NATS subject; there is no bind/connect distinction (the broker mediates delivery via the reply-to inbox).

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	NATS endpoint, e.g. `"nats://127.0.0.1:4222/ctrl"`.	required
`sample_type`	`int`	Wire encoding of frames sent by this socket. One of :data:`CI32`, :data:`CF64` (default), :data:`CF128`. The reply frame's type is determined by the :class:`Replier`.	`...`

Raises:

Type	Description
`ValueError`	If `sample_type` is not one of the three supported values.
`RuntimeError`	If `nats-server` isn't reachable at `endpoint`.

Examples:

>>> from doppler.stream import Requester, CF64
>>> req = Requester("nats://127.0.0.1:4222/t19116", CF64)
>>> type(req).__name__
'Requester'
>>> req.close()

Context-manager form:

>>> with Requester("nats://127.0.0.1:4222/t19117", CF64) as req:
...     type(req).__name__
'Requester'

Full REQ/REP round-trip (requires a live :class:Replier):

>>> import numpy as np, time
>>> from doppler.stream import Replier
>>> rep = Replier("nats://127.0.0.1:4222/ctrl", CF64)
>>> req = Requester("nats://127.0.0.1:4222/ctrl", CF64)
>>> time.sleep(0.05)
>>> req.send(np.ones(4, dtype=np.complex128),
...          sample_rate=int(1e6))
>>> request, hdr = rep.recv(timeout_ms=2000)
>>> rep.send(request)
>>> reply, hdr2 = req.recv(timeout_ms=2000)
>>> req.close(); rep.close()

init ¶

__init__(endpoint: str, sample_type: int = ...) -> None

Create a Requester socket and connect to endpoint.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	NATS endpoint, e.g. `"nats://127.0.0.1:4222/ctrl"`.	required
`sample_type`	`int`	Wire encoding of frames sent by this socket: :data:`CI32`, :data:`CF64` (default), or :data:`CF128`.	`...`

enter ¶

__enter__() -> Requester

Return self for use as a context manager.

Examples:

>>> from doppler.stream import Requester, CF64
>>> with Requester("nats://127.0.0.1:4222/t19118", CF64) as req:
...     type(req).__name__
'Requester'

exit ¶

__exit__(*args: object) -> None

Call :meth:close on context-manager exit.

send ¶

send(
    samples: NDArray[Any],
    sample_rate: float = 0,
    center_freq: float = 0,
    timestamp_ns: int | None = None,
) -> None

Publish a request frame with this Requester's reply-to inbox.

Use strictly alternating with :meth:recv (send, recv, send, recv, ...) — sending again before consuming the previous reply works mechanically but leaves an unread message in the inbox for the next :meth:recv to pick up, which will desync the request/reply pairing.

Parameters:

Name	Type	Description	Default
`samples`	`ndarray`	C-contiguous array whose dtype must match the socket's `sample_type`: `numpy.complex128` for :data:`CF64`, `numpy.clongdouble` for :data:`CF128`, `numpy.int32` for :data:`CI32`.	required
`sample_rate`	`float`	Samples per second written into the header (default 0).	`0`
`center_freq`	`float`	Centre frequency in Hz written into the header (default 0).	`0`
`timestamp_ns`	`int`	UNIX nanoseconds to stamp on `timestamp_ns` instead of a fresh `CLOCK_REALTIME` read (default `None`: auto-stamp).	`None`

Raises:

Type	Description
`TypeError`	If `samples.dtype` does not match the socket's `sample_type`.
`ValueError`	If `samples` is not C-contiguous.
`RuntimeError`	If the publish fails (e.g. the broker connection dropped).

Examples:

>>> from doppler.stream import Requester, CF64
>>> import numpy as np
>>> req = Requester("nats://127.0.0.1:4222/ctrl", CF64)
>>> req.send(np.ones(4, dtype=np.complex128),
...          sample_rate=int(1e6))

recv ¶

recv(
    timeout_ms: int = -1,
) -> Tuple[NDArray[Any], Dict[str, Any]]

Receive the reply frame from the :class:Replier.

Waits on this Requester's reply-to inbox. Call after :meth:send; calling without a prior send just blocks (or times out) on an empty inbox rather than raising immediately.

Parameters:

Name	Type	Description	Default
`timeout_ms`	`int`	Milliseconds to wait before raising :exc:`TimeoutError`. `-1` (default) blocks indefinitely.	`-1`

Returns:

Name	Type	Description
`samples`	`ndarray`	Decoded reply data. dtype mirrors the :class:`Replier`'s `sample_type`.
`header`	`dict`	Decoded `dp_header_t` fields — see :meth:`Subscriber.recv` for the full key list.

Raises:

Type	Description
`TimeoutError`	If `timeout_ms` elapses before a reply arrives (including calling `recv` without a prior `send`).

Examples:

>>> from doppler.stream import Requester, CF64
>>> req = Requester("nats://127.0.0.1:4222/ctrl", CF64)
>>> req.send(np.ones(4, dtype=np.complex128))
>>> reply, hdr = req.recv(timeout_ms=2000)

close ¶

close() -> None

Destroy the underlying NATS handle and release all resources.

Calls dp_req_destroy(). Safe to call multiple times.

Examples:

>>> from doppler.stream import Requester, CF64
>>> req = Requester("nats://127.0.0.1:4222/t19119", CF64)
>>> req.close()
>>> req.close()  # idempotent — no error

Replier ¶

NATS request/reply — receives a request frame, then sends a reply.

Wraps dp_rep_t. Subscribes to the endpoint's subject; each :meth:recv captures the request's reply-to inbox, and the next :meth:send publishes the reply directly to that inbox. Calling send before recv has captured a request raises :exc:RuntimeError immediately (there is no reply-to target yet) — unlike ZMQ's REP FSM, recv itself has no such restriction and simply waits for the next request.

Complements :class:Requester. Use for control-plane responses or signal-frame RPC where the Replier processes each frame and returns a result synchronously.

Both ends address the same NATS subject; there is no bind/connect distinction (the broker mediates delivery via the reply-to inbox).

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	NATS endpoint, e.g. `"nats://127.0.0.1:4222/ctrl"`.	required
`sample_type`	`int`	Wire encoding of frames sent by this socket (the reply). One of :data:`CI32`, :data:`CF64` (default), :data:`CF128`. The request frame's type is determined by the :class:`Requester`.	`...`

Raises:

Type	Description
`ValueError`	If `sample_type` is not one of the three supported values.
`RuntimeError`	If `nats-server` isn't reachable at `endpoint`.

Examples:

>>> from doppler.stream import Replier, CF64
>>> rep = Replier("nats://127.0.0.1:4222/t19120", CF64)
>>> type(rep).__name__
'Replier'
>>> rep.close()

Context-manager form:

>>> with Replier("nats://127.0.0.1:4222/t19121", CF64) as rep:
...     type(rep).__name__
'Replier'

Full REQ/REP server loop (requires a live :class:Requester):

>>> from doppler.stream import Requester
>>> import numpy as np, time
>>> rep = Replier("nats://127.0.0.1:4222/ctrl", CF64)
>>> req = Requester("nats://127.0.0.1:4222/ctrl", CF64)
>>> time.sleep(0.05)
>>> req.send(np.ones(4, dtype=np.complex128))
>>> request, hdr = rep.recv(timeout_ms=2000)
>>> rep.send(request, sample_rate=hdr["sample_rate"])
>>> reply, _ = req.recv(timeout_ms=2000)
>>> req.close(); rep.close()

init ¶

__init__(endpoint: str, sample_type: int = ...) -> None

Create a Replier socket and bind to endpoint.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	NATS endpoint, e.g. `"nats://127.0.0.1:4222/ctrl"`.	required
`sample_type`	`int`	Wire encoding of reply frames sent by this socket: :data:`CI32`, :data:`CF64` (default), or :data:`CF128`.	`...`

enter ¶

__enter__() -> Replier

Return self for use as a context manager.

Examples:

>>> from doppler.stream import Replier, CF64
>>> with Replier("nats://127.0.0.1:4222/t19122", CF64) as rep:
...     type(rep).__name__
'Replier'

exit ¶

__exit__(*args: object) -> None

Call :meth:close on context-manager exit.

recv ¶

recv(
    timeout_ms: int = -1,
) -> Tuple[NDArray[Any], Dict[str, Any]]

Receive one request frame from a :class:Requester.

Blocks until a request arrives or the timeout expires. Captures the request's reply-to inbox so the next :meth:send reaches the right Requester.

Parameters:

Name	Type	Description	Default
`timeout_ms`	`int`	Milliseconds to wait before raising :exc:`TimeoutError`. `-1` (default) blocks indefinitely.	`-1`

Returns:

Name	Type	Description
`samples`	`ndarray`	Decoded request data. dtype mirrors the :class:`Requester`'s `sample_type`.
`header`	`dict`	Decoded `dp_header_t` fields — see :meth:`Subscriber.recv` for the full key list.

Raises:

Type	Description
`TimeoutError`	If `timeout_ms` elapses before a request arrives.
`RuntimeError`	If the underlying NATS recv fails for any other reason.

Examples:

>>> from doppler.stream import Replier, CF64
>>> rep = Replier("nats://127.0.0.1:4222/ctrl", CF64)
>>> request, hdr = rep.recv(timeout_ms=5000)
>>> rep.send(request,
...          sample_rate=hdr["sample_rate"])

send ¶

send(
    samples: NDArray[Any],
    sample_rate: float = 0,
    center_freq: float = 0,
    timestamp_ns: int | None = None,
) -> None

Publish the reply to the request's captured reply-to inbox.

Must be called after :meth:recv; calling without a prior recv raises :exc:RuntimeError immediately (there is no reply-to target yet). After this call returns, the Replier is ready to recv the next request.

Parameters:

Name	Type	Description	Default
`samples`	`ndarray`	C-contiguous array whose dtype must match the socket's `sample_type`: `numpy.complex128` for :data:`CF64`, `numpy.clongdouble` for :data:`CF128`, `numpy.int32` for :data:`CI32`.	required
`sample_rate`	`float`	Samples per second written into the reply header (default 0).	`0`
`center_freq`	`float`	Centre frequency in Hz written into the reply header (default 0).	`0`
`timestamp_ns`	`int`	UNIX nanoseconds to stamp on `timestamp_ns` instead of a fresh `CLOCK_REALTIME` read (default `None`: auto-stamp).	`None`

Raises:

Type	Description
`TypeError`	If `samples.dtype` does not match the socket's `sample_type`.
`ValueError`	If `samples` is not C-contiguous.
`RuntimeError`	If `send` is called before `recv` has captured a request, or the publish otherwise fails.

Examples:

>>> from doppler.stream import Replier, CF64
>>> import numpy as np
>>> rep = Replier("nats://127.0.0.1:4222/ctrl", CF64)
>>> request, hdr = rep.recv(timeout_ms=5000)
>>> rep.send(np.zeros_like(request),
...          sample_rate=hdr["sample_rate"])

close ¶

close() -> None

Destroy the underlying NATS handle and release all resources.

Calls dp_rep_destroy(). Safe to call multiple times.

Examples:

>>> from doppler.stream import Replier, CF64
>>> rep = Replier("nats://127.0.0.1:4222/t19123", CF64)
>>> rep.close()
>>> rep.close()  # idempotent — no error

Helpers¶

get_timestamp_ns ¶

get_timestamp_ns() -> int

Current wall-clock time in nanoseconds since the UNIX epoch.

Calls clock_gettime(CLOCK_REALTIME) in the C layer. Useful for stamping outgoing frames when the caller does not supply its own timestamp_ns, or for computing round-trip latency from the value returned in the received header dict.

Returns:

Type	Description
`int`	Non-negative nanosecond timestamp. Guaranteed to be monotonically non-decreasing within a single process on any POSIX system that supports `CLOCK_REALTIME`.

Examples:

>>> from doppler.stream import get_timestamp_ns
>>> t = get_timestamp_ns()
>>> isinstance(t, int)
True
>>> t > 0
True

Gallery — Waveform I/O — One Capture, Four File Types Design — Telemetry — zero-cost scalar taps for running pipelines

Python Streaming API¶

Socket patterns¶

PUB / SUB — broadcast¶

PUSH / PULL — pipeline¶

REQ / REP — request/response¶

Timestamps¶

PUB / SUB¶

Publisher ¶

__init__ ¶

__enter__ ¶

__exit__ ¶

send ¶

close ¶

Subscriber ¶

__init__ ¶

__enter__ ¶

__exit__ ¶

recv ¶

close ¶

PUSH / PULL¶

Push ¶

__init__ ¶

__enter__ ¶

__exit__ ¶

send ¶

close ¶

Pull ¶

__init__ ¶

__enter__ ¶

__exit__ ¶

recv ¶

ack ¶

close ¶

REQ / REP¶

Requester ¶

__init__ ¶

__enter__ ¶

__exit__ ¶

send ¶

recv ¶

close ¶

Replier ¶

__init__ ¶

__enter__ ¶

__exit__ ¶

recv ¶

send ¶

close ¶

Helpers¶

get_timestamp_ns ¶

Related pages¶

init ¶

enter ¶

exit ¶

init ¶

enter ¶

exit ¶

init ¶

enter ¶

exit ¶

init ¶

enter ¶

exit ¶

init ¶

enter ¶

exit ¶

init ¶

enter ¶

exit ¶