savant_rs.py.utils

ZeroMQ utilities.

class savant_rs.py.utils.zeromq.AsyncZeroMQSource(socket: str, receive_timeout: int = 1000, receive_hwm: int = 50, source_id: str | None = None, source_id_prefix: str | None = None, set_ipc_socket_permissions: int | None = 511, blacklist_size: int = 1024, blacklist_ttl: int = 10)

Async ZeroMQ Source class.

async next_message() → ZeroMQMessage | None: Try to receive next message.

reader: NonBlockingReader

class savant_rs.py.utils.zeromq.BaseZeroMQSource(socket: str, receive_timeout: int = 1000, receive_hwm: int = 50, source_id: str | None = None, source_id_prefix: str | None = None, set_ipc_socket_permissions: int | None = 511, blacklist_size: int = 1024, blacklist_ttl: int = 10)

Base ZeroMQ Source class.

Parameters:

socket – zmq socket endpoint
receive_timeout – receive timeout socket option
receive_hwm – high watermark for inbound messages
source_id – filter inbound messages by source ID
source_id_prefix – filter inbound messages by topic prefix

property is_started

abstractmethod next_message() → ZeroMQMessage | None: Try to receive next message.

receiver: BlockingReader | NonBlockingReader

start(): Start ZeroMQ source.

terminate(): Finish and free zmq socket.

class savant_rs.py.utils.zeromq.Defaults

BLACKLIST_SIZE = 1024

BLACKLIST_TTL = 10

RECEIVE_HWM = 50

RECEIVE_RETRIES = 3

RECEIVE_TIMEOUT = 1000

SENDER_RECEIVE_TIMEOUT = 5000

SEND_HWM = 50

class savant_rs.py.utils.zeromq.ReceiverSocketTypes(*values)

Receiver socket types.

REP = Rep

ROUTER = Router

SUB = Sub

class savant_rs.py.utils.zeromq.SenderSocketTypes(*values)

Sender socket types.

DEALER = Dealer

PUB = Pub

REQ = Req

class savant_rs.py.utils.zeromq.ZeroMQMessage(topic, message, content)

content: bytes | None: Alias for field number 2

message: Message: Alias for field number 1

topic: List[int]: Alias for field number 0

class savant_rs.py.utils.zeromq.ZeroMQSource(socket: str, receive_timeout: int = 1000, receive_hwm: int = 50, source_id: str | None = None, source_id_prefix: str | None = None, set_ipc_socket_permissions: int | None = 511, blacklist_size: int = 1024, blacklist_ttl: int = 10)

ZeroMQ Source class.

next_message() → ZeroMQMessage | None: Try to receive next message.

reader: BlockingReader

savant_rs.py.utils.zeromq.get_zmq_socket_uri_options(uri: str) → str | None