HumeStreamClient

class HumeStreamClient(ClientBase)

Example

import asyncio

from hume import HumeStreamClient
from hume.models.config import BurstConfig
from hume.models.config import ProsodyConfig

async def main():
    client = HumeStreamClient("<your-api-key>")
    configs = [BurstConfig(), ProsodyConfig()]
    async with client.connect(configs) as socket:
        result = await socket.send_file("<your-audio-filepath>")
        print(result)

asyncio.run(main())

__init__

Construct a HumeStreamClient.

def __init__(api_key: str,
             *args: Any,
             open_timeout: Optional[int] = 10,
             close_timeout: Optional[int] = 10,
             **kwargs: Any)

Arguments:

• api_key str - Hume API key.
• open_timeout Optional[int] - Time in seconds before canceling socket open operation.
• close_timeout Optional[int] - Time in seconds before canceling socket close operation.

get_api_type

@classmethod
def get_api_type(cls) -> ApiType

Get the ApiType of the client.

Returns:

• ApiType - API type of the client.

connect

@asynccontextmanager
async def connect(
        configs: List[ModelConfigBase],
        stream_window_ms: Optional[int] = None) -> AsyncIterator[StreamSocket]

Connect to the streaming API.

Note: Only one config per model type should be passed.
If more than one config is passed for a given model type, only the last config will be used.

Arguments:

• configs List[ModelConfigBase] - List of job configs.
• stream_window_ms Optional[int] - Length of the sliding window in milliseconds to use when
  aggregating media across streaming payloads within one websocket connection.

StreamSocket

Streaming socket connection

class StreamSocket()

__init__

def __init__(protocol: "WebSocketClientProtocol",
             configs: List[ModelConfigBase],
             stream_window_ms: Optional[int] = None)

Construct a StreamSocket.

Arguments:

• protocol WebSocketClientProtocol - Protocol instance from websockets library.
• configs List[ModelConfigBase] - List of model configurations.
• stream_window_ms Optional[int] - Length of the sliding window in milliseconds to use when
  aggregating media across streaming payloads within one websocket connection.

Raises:

• HumeClientException - If there is an error processing media over the socket connection.

send_file

Send a file on the StreamSocket.

async def send_file(filepath: Union[str, Path]) -> Any

Arguments:

• filepath Path - Path to media file to send on socket connection.

Returns:

• Any - Response from the streaming API.

send_bytes

Send raw bytes on the StreamSocket.

async def send_bytes(bytes_data: bytes) -> Any

Note: Input should be base64 encoded bytes.
You can use base64.b64encode() to encode a raw string.

Arguments:

• bytes_data bytes - Raw bytes of media to send on socket connection.

Returns:

• Any - Response from the streaming API.

send_text

Send text on the StreamSocket.

async def send_text(text: str) -> Any

Note: This method is intended for use with a LanguageConfig.
When the socket is configured for other modalities this method will fail.

Arguments:

• text str - Text to send to the language model.

Raises:

• HumeClientException - If the socket is configured with a modality other than language.

Returns:

• Any - Response from the streaming API.

send_facemesh

Send facemesh landmarks on the StreamSocket.

async def send_facemesh(landmarks: List[List[List[float]]]) -> Any

Note: This method is intended for use with a FacemeshConfig.
When the socket is configured for other modalities this method will fail.

Arguments:

• landmarks List[List[List[float]]] - List of landmark points for multiple faces.
  The shape of this 3-dimensional list should be (n, 478, 3) where n is the number
  of faces to be processed, 478 is the number of MediaPipe landmarks per face and 3
  represents the (x, y, z) coordinates of each landmark.

Raises:

• HumeClientException - If the socket is configured with a modality other than facemesh.

Returns:

• Any - Response from the streaming API.

reset_stream

async def reset_stream() -> Any

Reset the streaming sliding window.

A sliding window of context is maintained for the lifetime of your streaming connection.
Call this method when some media has been fully processed and you want to continue using the same
streaming connection without leaking context across media samples.

Returns:

• Any - Response from the streaming API.

get_job_details

async def get_job_details() -> Any

Get details associated with the current streaming connection.

Returns:

• Any - Response from the streaming API.