Hume Expression Measurement Python SDK

The official Python client for the Hume Expression Measurement API. The API measures emotional expression in audio and images. Upload a file to receive every measurement in one response, or stream audio or JPEG images over a WebSocket to receive measurements while the media is still arriving.

The SDK provides synchronous and asynchronous clients for the upload, realtime, and run endpoints, typed models for every request, response, and message, and API key authentication.

1. Documentation
2. Python quickstart
3. Audio upload guide
4. Video upload guide
5. Audio realtime guide
6. Video realtime guide
7. Runs guide
8. API reference

Requirements

Python 3.10 or later.

Installation

pip install hume-expression-measurement

Authentication

Every request and every WebSocket connection carries your API key in the X-Hume-Api-Key header. Pass the key to the client, or set the HUME_API_KEY environment variable and construct the client without arguments.

from hume_expression_measurement import ExpressionMeasurementClient

client = ExpressionMeasurementClient(api_key="YOUR_API_KEY")

Keep API keys on a server. To use the API from a web page, relay through your server. See Authentication.

Preparing audio

The audio endpoints accept 16-bit PCM at 16 kHz, mono. The upload endpoint takes it as a WAV file or as headerless little-endian samples, and the realtime endpoint takes headerless samples only. Convert a recording with ffmpeg:

ffmpeg -i recording.wav -f s16le -acodec pcm_s16le -ac 1 -ar 16000 speech.pcm

To convert in Python instead, or to stream from a microphone, use the audio helpers.

Quickstart: upload

The upload endpoints take media you already have and return every measurement in one response.

Audio

from hume_expression_measurement import ExpressionMeasurementClient

client = ExpressionMeasurementClient()

with open("speech.pcm", "rb") as audio:
    response = client.audio.measure(file=("speech.pcm", audio, "application/octet-stream"))

for measurement in response.measurements:
    top = ", ".join(f"{score.name} {score.probability:.2f}" for score in measurement.expressions[:3])
    print(f"utterance {measurement.utterance_id}, {measurement.audio_start_ms} to {measurement.audio_end_ms} ms: {top}")

The tuple gives the part a filename and a content type. application/octet-stream labels the file as headerless samples; to upload a WAV file, pass audio/wav. The response lists every utterance in utterances and every measurement in measurements, each with the utterance_id it belongs to. A request holds up to 25MB, a little over 13 minutes of audio. The audio upload guide covers the request, the response, and every error.

Video

from hume_expression_measurement import ExpressionMeasurementClient

client = ExpressionMeasurementClient()

with open("photo.jpg", "rb") as image:
    response = client.video.measure(file=[("photo.jpg", image, "image/jpeg")])

for result in response.measurements:
    for face in result.faces:
        if face.expressions is None:
            continue
        top = ", ".join(f"{score.name} {score.probability:.2f}" for score in face.expressions[:3])
        print(f"image {result.frame_id}, face {face.face_id} at {face.bbox}: {top}")

file is a list, so one request can carry several JPEG images. measurements holds one result per image, in the order sent, and frame_id is the image's position in the list. The server lists up to 32 detected faces and measures only the largest, 2 by default. A face it did not measure has face_id, expressions, and descriptions set to None, so the example skips it. The video upload guide covers image limits, detection settings, and tracking faces across requests.

Quickstart: realtime

The realtime endpoints take media that is still arriving, such as audio from a live microphone or images from a camera, and send measurements as they are produced. A socket from connect() ends when its connection does; to continue after a server restart or a dropped connection, see Reconnecting.

Audio

Stream the audio in frames at the rate it plays, print the results as they arrive, and close the session.

import asyncio

from hume_expression_measurement import (
    AsyncExpressionMeasurementClient,
    Error,
    SessionClose,
    AudioMeasurementResult,
    AudioSessionClosed,
)

# 3200 bytes is 100 ms of audio at 16 kHz, 16-bit, mono.
FRAME_SIZE = 3200


async def main() -> None:
    client = AsyncExpressionMeasurementClient()

    async with client.audio.connect() as socket:

        async def print_measurements() -> None:
            async for message in socket:
                if isinstance(message, AudioMeasurementResult):
                    top = ", ".join(f"{score.name} {score.probability:.2f}" for score in message.expressions[:3])
                    print(f"utterance {message.utterance_id}, {message.audio_start_ms} to {message.audio_end_ms} ms: {top}")
                elif isinstance(message, Error):
                    print(f"{message.code}: {message.message}")
                elif isinstance(message, AudioSessionClosed):
                    print(f"closed after {message.produced.measurements} measurements")
                    break

        receiver = asyncio.create_task(print_measurements())
        try:
            with open("speech.pcm", "rb") as audio:
                while frame := audio.read(FRAME_SIZE):
                    await socket.send_audio_frame(frame)
                    await asyncio.sleep(0.1)
            await socket.send_audio_session_close(SessionClose(type="session.close"))
            await receiver
        finally:
            receiver.cancel()


asyncio.run(main())

The server detects speech, groups it into utterances, and sends a measurement.result every 3 seconds of speech while an utterance continues. It takes audio in close to real time, no faster than 1 second of audio per second, so the example waits 100 ms after each 100 ms frame. To measure a recording faster than it plays, upload it instead. The audio realtime guide covers the send rate, utterances, and the message flow.

Video

The video endpoint accepts one complete JPEG image per frame and answers each accepted frame with one measurement.result listing up to 32 detected faces.

from hume_expression_measurement import (
    Error,
    ExpressionMeasurementClient,
    SessionClose,
    VideoMeasurementResult,
    VideoSessionClosed,
)

client = ExpressionMeasurementClient()

with client.video.connect() as socket:
    with open("photo.jpg", "rb") as image:
        socket.send_image_frame(image.read())
    socket.send_video_session_close(SessionClose(type="session.close"))

    for message in socket:
        if isinstance(message, VideoMeasurementResult):
            for face in message.faces:
                if face.expressions is None:
                    continue
                top = ", ".join(f"{score.name} {score.probability:.2f}" for score in face.expressions[:3])
                print(f"face {face.face_id} at {face.bbox}: {top}")
        elif isinstance(message, Error):
            print(f"{message.code}: {message.message}")
        elif isinstance(message, VideoSessionClosed):
            break

bbox is [x0, y0, x1, y1] in pixels of the submitted image. face_id links the same face across frames within a session. The video realtime guide covers frame limits, the send rate, and face tracking. To stream a video file or a live source within the send rate, use the video helpers.

Handling messages

Iterating a socket yields one message at a time, each parsed into the model for its type. Check the model with isinstance, which also lets a type checker narrow the message to that model's fields. recv() returns the next message when you would rather pull one at a time. Every model is exported from hume_expression_measurement.

Async client

AsyncExpressionMeasurementClient has the same interface with async methods and iteration. The realtime audio quickstart uses it to read results while sending audio.

import asyncio

from hume_expression_measurement import AsyncExpressionMeasurementClient


async def main() -> None:
    client = AsyncExpressionMeasurementClient()

    with open("speech.pcm", "rb") as audio:
        response = await client.audio.measure(file=("speech.pcm", audio, "application/octet-stream"))
    print(f"{len(response.measurements)} measurements")


asyncio.run(main())

Audio helpers

hume_expression_measurement.audio_helpers records from a microphone and converts audio at any sample rate from 1 kHz and any channel count into frames for the realtime audio endpoint. Install it with the audio extra:

pip install "hume-expression-measurement[audio]"

The extra installs numpy, sounddevice, and soxr. soxr is licensed under the LGPL 2.1 or later and is installed only with this extra. On Linux, sounddevice also needs the PortAudio library, for example sudo apt-get install libportaudio2 on Debian and Ubuntu. If the extra is missing, importing hume_expression_measurement.audio_helpers or calling a helper raises MissingDependencyError, whose message gives the install command.

Microphone

Microphone.open() records from the default input device, or the one you pass as device, and yields 100 ms frames ready to send. This example records for 10 seconds while printing measurements as they arrive, then closes the session.

import asyncio

from hume_expression_measurement import (
    AsyncExpressionMeasurementClient,
    Error,
    SessionClose,
    AudioMeasurementResult,
    AudioSessionClosed,
)
from hume_expression_measurement.audio_helpers import Microphone


async def main() -> None:
    client = AsyncExpressionMeasurementClient()

    # Opening the microphone first means a missing or busy device fails before a session starts.
    async with Microphone.open() as microphone, client.audio.connect() as socket:

        async def print_measurements() -> None:
            async for message in socket:
                if isinstance(message, AudioMeasurementResult):
                    top = ", ".join(f"{score.name} {score.probability:.2f}" for score in message.expressions[:3])
                    print(f"utterance {message.utterance_id}, {message.audio_start_ms} to {message.audio_end_ms} ms: {top}")
                elif isinstance(message, Error):
                    print(f"{message.code}: {message.message}")
                elif isinstance(message, AudioSessionClosed):
                    break

        receiver = asyncio.create_task(print_measurements())
        try:
            print(f"Recording from {microphone.device_name} for 10 seconds")
            frames_sent = 0
            async for frame in microphone:
                await socket.send_audio_frame(frame)
                frames_sent += 1
                # 100 frames of 100 ms is 10 seconds.
                if frames_sent == 100:
                    break

            await socket.send_audio_session_close(SessionClose(type="session.close"))
            await receiver
        finally:
            receiver.cancel()


asyncio.run(main())

The device records at its own sample rate, and the SDK converts its audio to 16 kHz mono. Each async for yields only audio recorded after it starts, so a push-to-talk loop that starts a new async for for each press never sends audio recorded between presses. If the loop body is too slow and more than 10 seconds of audio builds up, iteration raises a RuntimeError. An unknown device, one that is not an input device, or a name that matches several devices raises a ValueError that lists the available input devices. On Windows, where each device is listed once per host API, add the host API to the name, for example "Microphone WASAPI", or pass the index. On macOS, the first recording asks permission for the terminal or app running Python; if access is denied, the device records silence.

WAV files

iter_wav_frames reads 8, 16, 24, or 32-bit integer and 32 or 64-bit floating-point WAV at any sample rate from 1 kHz and any channel count and yields 100 ms frames. It replaces the file loop in the realtime quickstart:

from hume_expression_measurement.audio_helpers import iter_wav_frames

for frame in iter_wav_frames("recording.wav"):
    await socket.send_audio_frame(frame)
    await asyncio.sleep(0.1)

Frames are produced as fast as the file is read, so the loop waits 100 ms after each one to stream the audio in close to real time. Compressed WAV files raise a ValueError with an ffmpeg command that converts them.

Other sources

Resampler converts a stream you already have, as bytes or NumPy arrays, block by block. Pass each block to process, then call flush after the last one. process can return an empty result while the filter fills, and the endpoint rejects empty frames, so check before sending. The endpoint also rejects frames longer than 10 seconds, so pass a long recording in blocks of about 100 ms rather than in one call, and stream the audio in close to real time, no faster than 1 second of audio per second.

from hume_expression_measurement.audio_helpers import Resampler

resampler = Resampler(sample_rate=48000, channels=2, sample_format="int16")
for block in blocks:
    if pcm := resampler.process(block):
        socket.send_audio_frame(pcm)
if pcm := resampler.flush():
    socket.send_audio_frame(pcm)

Video helpers

hume_expression_measurement.video_helpers streams JPEG frames to the realtime video endpoint with the async client. stream_video sends frames within the send rate, handles rejected frames, and pairs every reply with the frame it answers. iter_video_frames reads frames from a video file through ffmpeg, and JpegSplitter splits a stream of concatenated JPEG images, such as MJPEG, into single frames. The helpers need no extra.

Streaming frames

stream_video takes a socket and any async iterable of VideoFrames, each a JPEG image as data with its time in milliseconds as timestamp_ms, and yields one reply per frame, in the order of the frames. A reply holds the frame's timestamp_ms and either the result that measured it or the error that rejected it, with the other set to None. This example measures a video file as it plays:

import asyncio

from hume_expression_measurement import AsyncExpressionMeasurementClient
from hume_expression_measurement.video_helpers import iter_video_frames, stream_video


async def main() -> None:
    client = AsyncExpressionMeasurementClient()

    async with client.video.connect() as socket:
        stream = stream_video(socket, iter_video_frames("interview.mp4", fps=2, realtime=True))
        async for reply in stream:
            if reply.result is not None:
                for face in reply.result.faces:
                    if face.expressions is None:
                        continue
                    top = ", ".join(f"{score.name} {score.probability:.2f}" for score in face.expressions[:3])
                    print(f"{reply.timestamp_ms} ms, face {face.face_id}: {top}")
            else:
                print(f"{reply.timestamp_ms} ms: {reply.error.code}")
        if stream.closed is not None:
            print(f"{stream.closed.produced.measurements} faces measured")


asyncio.run(main())

realtime=True produces the frames as the video plays, as the endpoint expects. Without it, stream_video would send these frames, taken at 2 per second, at its full rate of 3 per second, faster than the video plays.

stream_video sends frames no faster than 3 per second, the endpoint's send rate, counted from when streaming starts. A frame that arrives late leaves room for the next one to arrive early by as much, up to a third of a second, and still be sent rather than held back or dropped. When the server rejects a frame with rate_limited, stream_video halves its send rate, down to one frame every 2 seconds, and raises it again gradually while frames are accepted, so a tighter limit than expected costs only a few rejected frames. What happens to the rejected frame depends on the source:

1. From a recording, the default, it sends the frame again once the lower rate allows, so every frame is measured. It also sends a frame once more after a retryable internal_error.
2. From a live source, with live=True, it yields the rejection and moves on, so results keep up with the source. A live frame that arrives when the send rate has no room is not sent at all, and stream.dropped_frames counts it.

A frame over the 2 MB message limit would end the session, so it is never sent and is yielded as a message_too_large error. Frames that are not valid JPEGs or that exceed 8,294,400 pixels are rejected by the server with invalid_image_frame, and the session continues.

When the frames run out, stream_video waits for the remaining replies, then closes the session, and the server closes the connection. stream.closed then holds the session.closed message with the session's totals. If the session ends before the frames do, because of a session-ending error, an idle timeout, or a dropped connection, the loop raises a SessionEndedError. Its closed holds the session.closed message when one arrived, and its code names the error that ended the session, if there was one. A session cannot be resumed, so each stream needs its own socket from client.video.connect(), and calling stream_video again with a socket it has already used raises a RuntimeError.

Leaving the loop early, or an error from the frames, ends the session and stops the frames. After a break, stream_video cleans up once the stream's iterator is garbage collected, and if the async with block has closed the connection by then, the closed connection is what ends the session. To clean up as soon as the block exits, wrap the iterator in contextlib.aclosing:

async with contextlib.aclosing(aiter(stream_video(socket, frames))) as replies:
    async for reply in replies:
        ...

stream_video reads every message from the socket while it runs, so read messages from the replies instead. The configuration locks at the first frame, so send any session.update before streaming. Frames are sent only while the loop asks for the next reply, so keep the loop body short.

Video files

iter_video_frames runs ffmpeg, which must be installed: brew install ffmpeg on macOS, sudo apt install ffmpeg on Debian or Ubuntu, or winget install Gyan.FFmpeg on Windows. It reads any format ffmpeg can decode and yields frames whose timestamp_ms is their position in the video. If ffmpeg is not on PATH or at ffmpeg_path, iteration raises an FfmpegNotFoundError, a FileNotFoundError whose message says how to install ffmpeg. If ffmpeg fails, iteration raises a RuntimeError that includes ffmpeg's output. ffmpeg starts when the first frame is requested, so with stream_video these errors are raised from the loop after the session has started, and stream_video has asked the server to close the session before they reach you.

stream_video stops ffmpeg when the stream ends. If you iterate iter_video_frames yourself, leaving the loop early stops ffmpeg once the generator is garbage collected, or as soon as the block exits if you wrap the generator in contextlib.aclosing.

Live sources

stream_video measures frames from any source you can turn into an async iterable, such as a camera your application already captures. Encode each frame as a JPEG, give it a timestamp_ms, and pass live=True. A camera usually produces more frames than the 3 per second the endpoint accepts; stream_video sends what the send rate allows and counts the rest in stream.dropped_frames, so capturing no more than 3 frames per second saves encoding work. A source that keeps to exactly 3 frames per second can still lose one frame: the first that would put it ahead of 3 per second, counted from when streaming started. That drop leaves a frame of room, so after it a frame is dropped only if the source's timing varies by more than about a sixth of a second. Keep frames around 1280 pixels wide.

JpegSplitter turns a byte stream of concatenated JPEG images, such as ffmpeg's image2pipe output or the body of an MJPEG stream, into single images. Pass each chunk to push, which returns the images completed so far and holds back a partial one. This generator yields frames from such a stream, timed from when it starts:

import time
import typing

from hume_expression_measurement.video_helpers import JpegSplitter, VideoFrame, stream_video


async def frames_from(mjpeg: typing.AsyncIterable[bytes]) -> typing.AsyncIterator[VideoFrame]:
    splitter = JpegSplitter()
    start = time.monotonic()
    async for chunk in mjpeg:
        for data in splitter.push(chunk):
            yield VideoFrame(data=data, timestamp_ms=round((time.monotonic() - start) * 1000))


async for reply in stream_video(socket, frames_from(mjpeg), live=True):
    print(reply.timestamp_ms, reply.error.code if reply.error is not None else len(reply.result.faces))

Scores

Every measurement contains lists of scores, each pairing a name with a probability from 0 to 1. Scores are sorted by descending probability, and each list includes only the scores that pass its cutoff, so lists vary in length and may be empty. Audio results carry expressions, each included when judged present against a threshold set for that expression, and voice_attributes, included at 0.725 or above. Video results carry expressions and descriptions for each face, both included above 0.1. What probability denotes differs from list to list. Scores explains how to read them.

Configuration

These settings apply to both upload and realtime requests.

On an upload, pass the settings as config:

from hume_expression_measurement import AudioFileConfig, VideoFileConfig, VideoFileConfigFace

response = client.audio.measure(file=audio_file, config=AudioFileConfig(measurement_timer_ms=5000))
response = client.video.measure(file=images, config=VideoFileConfig(face=VideoFileConfigFace(threshold=0.8, min_size=40)))

In a session, send session.update before the first frame. The server replies with session.updated. The configuration locks once the first frame is accepted, and a field omitted from session.update returns to its default.

from hume_expression_measurement import AudioSessionUpdate

socket.send_audio_session_update(AudioSessionUpdate(type="session.update", measurement_timer_ms=5000))

from hume_expression_measurement import FaceConfigUpdate, VideoSessionUpdate

socket.send_video_session_update(
    VideoSessionUpdate(type="session.update", face=FaceConfigUpdate(threshold=0.8, min_size=40))
)

Reconnecting

reconnecting from hume_expression_measurement.reconnect connects in place of connect() and yields a socket with the same send methods, recv(), and iteration. When the connection ends unexpectedly, the socket connects again, which starts a new session. It reconnects after:

1. A session.closed with reason set to server_shutdown, which the server sends when it shuts down, as during a deployment.
2. The connection closing with code 1001, 1011, 1012, or 1013, or ending without a close frame, as when it drops.
3. An audio internal_error, after which the API asks for a new connection.

Any other close is final, and so is any close after you send session.close or leave the block. Receiving is what notices a close and reconnects, so keep iterating or calling recv() while the session runs. This example measures the microphone until the socket stops reconnecting:

import asyncio

from hume_expression_measurement import (
    AsyncExpressionMeasurementClient,
    AudioMeasurementResult,
    AudioSessionCreated,
    AudioSessionUpdate,
)
from hume_expression_measurement.audio_helpers import Microphone
from hume_expression_measurement.reconnect import reconnecting
from websockets.exceptions import ConnectionClosed


async def main() -> None:
    client = AsyncExpressionMeasurementClient()

    async with Microphone.open() as microphone, reconnecting(client.audio) as socket:
        await socket.send_audio_session_update(AudioSessionUpdate(type="session.update", measurement_timer_ms=5000))

        async def send_audio() -> None:
            async for frame in microphone:
                try:
                    await socket.send_audio_frame(frame)
                except ConnectionClosed:
                    # Audio recorded while the socket reconnects is skipped.
                    continue

        sender = asyncio.create_task(send_audio())
        try:
            async for message in socket:
                if isinstance(message, AudioSessionCreated):
                    print(f"session {message.session_id}")
                elif isinstance(message, AudioMeasurementResult):
                    top = ", ".join(f"{score.name} {score.probability:.2f}" for score in message.expressions[:3])
                    print(f"utterance {message.utterance_id}: {top}")
        finally:
            sender.cancel()


asyncio.run(main())

With ExpressionMeasurementClient, enter reconnecting(client.audio) with with. Its socket accepts sends from other threads while one thread receives, so send from one thread and iterate in another.

The first attempt waits 1 to 5 seconds, and each later attempt 1.3 times as long as the one before, up to 10 seconds. An attempt that fails with a network error, a timeout, or a 408, 429, or 5xx status is retried, up to 30 attempts in a row, and the count starts again once a connection has stayed open for 5 seconds. When the 30 attempts run out, receiving raises the last attempt's error. If that attempt connected but closed again within 5 seconds, receiving raises a websockets.exceptions.ConnectionClosedError for that close, even a normal one, so that iteration does not end as though the session had finished. An attempt refused with any other 4xx status, such as 401 for an API key that is no longer valid, is not retried, and receiving raises its error at once. A failed first connection is not retried, since a wrong URL or API key would fail the same way again: entering the block raises what connect() raises.

A new session does not continue the previous one. Every message is passed on, so you receive the previous session's session.closed, if it sent one, then session.created again with a new session ID, and so a new run. Keep receiving after a session.closed rather than stopping there as the examples for connect() do, since the socket reconnects only while you receive. Once the socket has ended for good, iteration ends or raises on its own. IDs and times in the new session's results start again from 0. Results for media that the previous session had received but not yet measured never arrive. The new session receives your last session.update before anything else, and the server confirms it with session.updated as usual.

While the socket reconnects, sending raises websockets.exceptions.ConnectionClosed and has no effect, except that session.close stops the reconnect. A live source can skip frames until sending succeeds again, as the example does. A recording loses whatever the previous session had not measured, so rather than skip frames, let the error end the loop and leave the block, which stops the reconnect. Then send the recording again on a new connection or upload it.

stream_video does not reconnect. It raises a SessionEndedError when the session or the connection ends before the frames do, and a TypeError when given a socket from reconnecting.

Runs

Every upload request that is measured and every realtime session is recorded as a run. A run's ID is the run_id of an upload response or the session_id of a session. The run endpoints return a run's status, configuration, and event log. They do not return measurements, so keep the results you need from the upload response or the session.

from hume_expression_measurement import ExpressionMeasurementClient

client = ExpressionMeasurementClient()

page = client.runs.list(endpoint="audio", mode="realtime")
for summary in page.runs:
    print(summary.run_id, summary.status, summary.metered_seconds)

run = client.runs.get("01926f3a-5b1c-7d2e-8f40-3a9b7c1d2e5f")
print(run.status, run.close_reason)

events = client.runs.list_events("01926f3a-5b1c-7d2e-8f40-3a9b7c1d2e5f")
for event in events.events:
    print(event.seq, event.kind)

runs.list and runs.list_events return one page at a time. Pass a page's next_cursor back as cursor to fetch the next one; the last page has none. The runs guide covers filters, run status, and every event kind.

Errors

The upload and run methods raise hume_expression_measurement.core.ApiError when a request fails. status_code holds the HTTP status, and body holds the error body. For every status except 401, body is an ErrorBody whose code says what went wrong; a 401 carries only a message.

from hume_expression_measurement.core import ApiError

try:
    response = client.video.measure(file=[("photo.jpg", image, "image/jpeg")])
except ApiError as error:
    print(f"{error.status_code}: {error.body}")

Problems during a session arrive as error messages rather than exceptions. A rejected frame, such as invalid_audio_frame, invalid_image_frame, or rate_limited, is discarded and the session continues. config_invalid, message_too_large, an audio internal_error, and five consecutive video internal_error messages end the session, and session.closed follows with reason set to error. When the server refuses the WebSocket handshake, connect() raises ApiError with the HTTP status in status_code and the server's response headers in headers. A status of 401 means the API key was rejected. A socket from reconnecting raises the same error when the server refuses a reconnect.

Errors lists every code and whether to retry.

Contributing

Most of this SDK is generated by Fern from the API definition, and edits to generated files are overwritten on the next generation. CONTRIBUTING.md explains how to build and test the project and how to add code that persists across regenerations. To report a problem, open an issue in this repository.

License

MIT