drava_common.publisher

Shared publisher helpers for Drava example data-source processes.

Publishers are not the runtime: they are plain NATS/socket clients that feed frames into stage1 and emit the end-of-stream (EOS) marker. Every example used to copy the same pacing loop, EOS emission, metrics writer, and config resolution. That lives here now; an example publisher only supplies a payload generator and picks the transport.

Config precedence matches the documented rule for publishers: an explicit env var wins, otherwise the value from pipeline.yaml (via DRAVA_STAGE_CONFIG), otherwise a built-in default.

Attributes

EOS_PREFIX

Functions

connect_jetstream(nats_url, stream, subject)	Connect to NATS and ensure stream exists carrying subject.
load_transport_config()	Return (nats_url, stream, subject) for feeding stage1.
load_publish_config([default_num_frames])	Return (rate_hz, synthetic_mode, num_frames).
write_publisher_metrics(frames, duration_s, avg_fps[, ...])	Write a single JSON metrics object to $DRAVA_PUBLISHER_METRICS_FILE.
publish_stream(js, subject, next_payload, num_frames)	Publish num_frames payloads to a JetStream subject then an EOS.
socket_publish_stream(fifo_path, next_payload, num_frames)	Publish num_frames payloads to a Drava socket FIFO then an EOS.

Module Contents

drava_common.publisher.EOS_PREFIX = b'DRAVA_EOS:'

async drava_common.publisher.connect_jetstream(nats_url: str, stream: str, subject: str)

Connect to NATS and ensure stream exists carrying subject.

Returns a JetStream context ready for publish_stream. Idempotent: an already-existing stream is fine. Keeps the NATS client alive on the returned context (._nc) so callers can await ctx._nc.drain() at the end.

drava_common.publisher.load_transport_config()

Return (nats_url, stream, subject) for feeding stage1.

Env vars NATS_URL / DRAVA_STREAM / DRAVA_SUBJECT override the corresponding YAML values (transport.nats_url and stage1’s ingress).

drava_common.publisher.load_publish_config(default_num_frames: int | None = None)

Return (rate_hz, synthetic_mode, num_frames).

• DRAVA_PUBLISH_RATE_HZ overrides publisher.rate_hz (default 0 = max).

• DRAVA_PUBLISH_SYNTHETIC overrides publisher.synthetic (default off).

• DRAVA_PUBLISH_NUM_FRAMES overrides publisher.num_frames. If neither is set, default_num_frames is used (e.g. a dataset size); if that is also None, raises.

drava_common.publisher.write_publisher_metrics(frames, duration_s, avg_fps, eos_seq=None)

Write a single JSON metrics object to $DRAVA_PUBLISHER_METRICS_FILE.

No-op when the env var is unset. Mirrors the runtime’s file-based metrics so orchestrators read files instead of scraping stdout.

async drava_common.publisher.publish_stream(js, subject: str, next_payload: Callable[[int], bytes], num_frames: int, rate_hz: float = 0.0, log_every: int = 1024, inflight: int = 1024, retries: int = 8, retry_delay_s: float = 0.05, drain: bool = True)

Publish num_frames payloads to a JetStream subject then an EOS.

js is a NATS JetStream context (e.g. from connect_jetstream()); next_payload(i) returns the bytes for frame i. Handles pacing, an in-flight publish window, retry-with-backoff on JetStream overflow, the end-of-stream marker, windowed FPS logging, and metrics.

Returns (sent_count, duration_s, avg_fps, eos_seq) and writes publisher metrics if $DRAVA_PUBLISHER_METRICS_FILE is set. If drain and the context exposes its NATS client (js._nc), the client is drained at the end so all acks land before exit.

drava_common.publisher.socket_publish_stream(fifo_path: str, next_payload: Callable[[int], bytes], num_frames: int, rate_hz: float = 0.0, log_every: int = 256)

Publish num_frames payloads to a Drava socket FIFO then an EOS.

Wire format per frame: [4-byte big-endian length][raw bytes]. Returns (sent_count, duration_s, avg_fps) and writes publisher metrics if set.