`snoozeml.streams` — Event Streaming Envelopes & Clients#

Event streaming message envelopes, wire serializers, and broker clients for SnoozeML.

Maintains strict binary and JSON wire parity with SnoozeML’s Rust streaming engine (src/stream/message.rs, src/stream/redis_stream.rs, src/stream/kafka.rs, and src/stream/rabbitmq.rs).

class snoozeml.streams.StreamWireFormat(*values)[source]#

Bases: str, Enum

Supported payload serialization formats across SnoozeML streaming transports.

JSON = 'json'#: Standard JSON serialization matching StreamInferRequest and StreamInferResponse.

PROTOBUF = 'protobuf'#: Binary Protobuf serialization matching snoozeml.v1.ModelInferRequest and ModelInferResponse.

classmethod from_str(value)[source]#

Parses a case-insensitive wire format string into a StreamWireFormat.

Parameters:: value (StreamWireFormat | str)
Return type:: StreamWireFormat

class snoozeml.streams.StreamInferRequest(inputs, model_id=None, id=<factory>, reply_to=None)[source]#

Bases: object

Outgoing streaming inference request envelope matching src/stream/message.rs.

Parameters:

inputs (List[TensorInput])
model_id (str | None)
id (str)
reply_to (str | None)

inputs#

Validated list of TensorInput tensors.

Type:: List[snoozeml.inference.TensorInput]

model_id#

Optional target model identifier (overrides the stream binding’s default model).

Type:: str | None

id#

Correlation identifier used to match responses to requests (auto-generated UUID if omitted).

Type:: str

reply_to#

Optional dynamic destination topic, queue, or stream key for routing the response.

Type:: str | None

inputs: List[TensorInput]#

model_id: str | None = None#

id: str#

reply_to: str | None = None#

classmethod from_inputs(inputs, *, model_id=None, id=None, reply_to=None, default_input_name='input')[source]#

Constructs a StreamInferRequest from NumPy arrays, PyTorch tensors, or dicts.

Parameters:

inputs (Mapping[str, Any | Sequence[Any] | TensorInput] | Sequence[TensorInput | Mapping[str, Any]] | TensorInput | Any) – Input tensors in any format accepted by normalize_inputs().
model_id (str | None) – Optional target model identifier.
id (str | None) – Optional explicit correlation ID (auto-generated if None).
reply_to (str | None) – Optional dynamic egress topic/queue override.
default_input_name (str) – Binding name when a bare array is provided.

Returns:

Validated StreamInferRequest.

Return type:

StreamInferRequest

to_dict()[source]#

Serializes the request envelope into a JSON-compatible dictionary.

Return type:: Dict[str, Any]

to_bytes(format=StreamWireFormat.JSON)[source]#

Encodes the request envelope into raw wire bytes for publishing to a broker.

Parameters:: format (StreamWireFormat | str) – Serialization format (StreamWireFormat.JSON or StreamWireFormat.PROTOBUF).
Returns:: Encoded byte payload ready for Kafka, RabbitMQ, or Redis Streams.
Return type:: bytes

classmethod from_bytes(payload, format=StreamWireFormat.JSON)[source]#

Deserializes a StreamInferRequest from raw broker bytes.

Parameters:

payload (bytes | bytearray | memoryview | str) – Raw bytes or UTF-8 string read from a stream broker.
format (StreamWireFormat | str) – Wire format (StreamWireFormat.JSON or StreamWireFormat.PROTOBUF).

Returns:

Decoded StreamInferRequest.

Return type:

StreamInferRequest

class snoozeml.streams.StreamInferResponse(id, model_id, status='success', outputs=<factory>, error=None, duration_ms=0.0)[source]#

Bases: object

Incoming streaming inference response envelope published by SnoozeML to output/DLQ topics.

Parameters:

id (str)
model_id (str)
status (str)
outputs (List[TensorOutput])
error (str | None)
duration_ms (float)

id#

Correlation ID matching the original StreamInferRequest.

Type:: str

model_id#

Identifier of the model that executed the request.

Type:: str

status#

Execution status ("success" or "error").

Type:: str

outputs#

Output tensors wrapped in an OutputList.

Type:: List[snoozeml.inference.TensorOutput]

error#

Diagnostic error message when status == "error".

Type:: str | None

duration_ms#

Server-side execution duration in milliseconds.

Type:: float

id: str#

model_id: str#

status: str = 'success'#

outputs: List[TensorOutput]#

error: str | None = None#

duration_ms: float = 0.0#

property is_success: bool#: Returns True if the stream inference succeeded (status == "success").

raise_for_status()[source]#

Raises a RuntimeInferenceError if the response represents an error.

Return type:: None

to_inference_response()[source]#

Converts a successful stream response into a standard InferenceResponse.

Raises:: RuntimeInferenceError – If is_success is False.
Return type:: InferenceResponse

get_tensor(name=None)[source]#

Retrieves an output TensorOutput by name (or the first output).

Parameters:: name (str | None)
Return type:: TensorOutput

get_numpy(name=None)[source]#

Returns the output tensor as a reshaped np.ndarray (dtype=np.float32).

Parameters:: name (str | None)
Return type:: ndarray

get_torch(name=None, device=None)[source]#

Returns the output tensor as a torch.Tensor.

Parameters:

name (str | None)
device (str | None)

Return type:

Any

to_dict()[source]#

Serializes the response envelope into a JSON-compatible dictionary.

Return type:: Dict[str, Any]

to_bytes(format=StreamWireFormat.JSON)[source]#

Encodes this response envelope into raw bytes matching src/stream/message.rs.

Parameters:: format (StreamWireFormat | str)
Return type:: bytes

classmethod from_dict(data)[source]#

Constructs a StreamInferResponse from a decoded JSON dictionary.

Parameters:: data (Mapping[str, Any])
Return type:: StreamInferResponse

classmethod from_bytes(payload, format=StreamWireFormat.JSON, default_id='')[source]#

Deserializes a StreamInferResponse from raw broker bytes.

Parameters:

payload (bytes | bytearray | memoryview | str) – Raw bytes or UTF-8 string read from an output or DLQ stream.
format (StreamWireFormat | str) – Wire format (StreamWireFormat.JSON or StreamWireFormat.PROTOBUF).
default_id (str) – Fallback correlation ID when decoding Protobuf responses.

Returns:

Decoded StreamInferResponse.

Return type:

StreamInferResponse

class snoozeml.streams.MockStreamClient(*, client=None, handler=None, default_model_id=None, input_topic='snooze.requests', output_topic='snooze.responses', error_topic='snooze.dlq', format=StreamWireFormat.JSON)[source]#

Bases: object

Deterministic in-memory streaming client for testing stream pipelines in Python.

Can be wired either to a live SnoozeClient or to a custom inference handler callback, simulating SnoozeML’s src/stream/mock_stream.rs broker with full JSON and Protobuf wire serialization round-trips.

Parameters:

client (Optional[Any])
handler (Optional[Callable[[str, List[TensorInput]], List[TensorOutput]]])
default_model_id (Optional[str])
input_topic (str)
output_topic (str)
error_topic (str)
format (Union[StreamWireFormat, str])

topics: Dict[str, Deque[bytes]]#

publish_request(inputs, *, model_id=None, request_id=None, reply_to=None, topic=None)[source]#

Serializes and publishes a StreamInferRequest onto the input topic.

If client or handler is attached, immediately processes the message and publishes the serialized StreamInferResponse onto reply_to or output_topic.

Returns:

Correlation ID of the published request.

Parameters:

inputs (Mapping[str, Any | Sequence[Any] | TensorInput] | Sequence[TensorInput | Mapping[str, Any]] | TensorInput | Any)
model_id (str | None)
request_id (str | None)
reply_to (str | None)
topic (str | None)

Return type:

str

process_next(*, topic=None)[source]#

Pops one raw request frame from topic, executes inference, and writes the response frame.

Parameters:: topic (str | None)
Return type:: StreamInferResponse | None

read_response(*, correlation_id=None, topic=None)[source]#

Reads and deserializes the next matching StreamInferResponse from topic.

Parameters:

correlation_id (str | None)
topic (str | None)

Return type:

StreamInferResponse | None

infer(inputs, *, model_id=None, reply_to=None)[source]#

Publishes a request and returns the resulting InferenceResponse.

Parameters:

inputs (Mapping[str, Any | Sequence[Any] | TensorInput] | Sequence[TensorInput | Mapping[str, Any]] | TensorInput | Any)
model_id (str | None)
reply_to (str | None)

Return type:

InferenceResponse

class snoozeml.streams.RedisStreamClient(endpoint='redis://127.0.0.1:6379', *, input_topic='snooze.requests', output_topic='snooze.responses', default_model_id=None, format=StreamWireFormat.JSON, redis_client=None)[source]#

Bases: object

Redis Streams producer/consumer client compatible with SnoozeML’s RedisStreamDriver.

Publishes requests via XADD <input_topic> * payload <bytes> and reads responses from <output_topic> (or per-request reply_to) via XREAD.

Parameters:

endpoint (str) – Redis connection URL (e.g. "redis://127.0.0.1:6379"). Ignored if redis_client is passed.
input_topic (str) – Redis stream key where SnoozeML consumes requests (default "snooze.requests").
output_topic (str) – Redis stream key where SnoozeML publishes responses (default "snooze.responses").
default_model_id (Optional[str]) – Optional default model_id attached to outgoing requests.
format (Union[StreamWireFormat, str]) – Wire serialization format ("json" or "protobuf").
redis_client (Optional[Any]) – Optional pre-configured redis.Redis client instance.

publish_request(inputs, *, model_id=None, request_id=None, reply_to=None, topic=None)[source]#

Publishes a StreamInferRequest to Redis Streams via XADD.

Returns:

Tuple of (correlation_id, redis_stream_entry_id).

Parameters:

inputs (Mapping[str, Any | Sequence[Any] | TensorInput] | Sequence[TensorInput | Mapping[str, Any]] | TensorInput | Any)
model_id (str | None)
request_id (str | None)
reply_to (str | None)
topic (str | None)

Return type:

Tuple[str, str]

wait_for_response(correlation_id, *, topic=None, last_id='0-0', timeout=10.0)[source]#

Polls the Redis output stream via XREAD until the matching correlation_id arrives.

Parameters:

correlation_id (str) – Request correlation ID to match.
topic (str | None) – Output stream key (defaults to output_topic).
last_id (str) – Redis stream cursor ID to start reading after (default "0-0").
timeout (float) – Maximum wait time in seconds before raising InferenceTimeoutError.

Returns:

Matching StreamInferResponse.

Return type:

StreamInferResponse

infer(inputs, *, model_id=None, reply_to=None, timeout=10.0)[source]#

Publishes a request to Redis Streams and waits for the correlated InferenceResponse.

Parameters:

inputs (Mapping[str, Any | Sequence[Any] | TensorInput] | Sequence[TensorInput | Mapping[str, Any]] | TensorInput | Any)
model_id (str | None)
reply_to (str | None)
timeout (float)

Return type:

InferenceResponse

snoozeml.streams — Event Streaming Envelopes & Clients#

`snoozeml.streams` — Event Streaming Envelopes & Clients#