Streaming
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.
Updated 3 months ago