Skip to content

HumeStreamClient

Bases: ClientBase

Streaming API client.

Example 
import asyncio

from hume import HumeStreamClient, StreamSocket
from hume.models.config import FaceConfig

async def main():
    client = HumeStreamClient("<your-api-key>")
    config = FaceConfig(identify_faces=True)
    async with client.connect([configs]) as socket:
        result = await socket.send_file("<your-image-filepath>")
        print(result)

asyncio.run(main())
Source code in hume/_stream/hume_stream_client.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class HumeStreamClient(ClientBase):
    """Streaming API client.

    Example:
        ```python
        import asyncio

        from hume import HumeStreamClient, StreamSocket
        from hume.models.config import FaceConfig

        async def main():
            client = HumeStreamClient("<your-api-key>")
            config = FaceConfig(identify_faces=True)
            async with client.connect([configs]) as socket:
                result = await socket.send_file("<your-image-filepath>")
                print(result)

        asyncio.run(main())
        ```
    """

    _DEFAULT_API_TIMEOUT = 10

    def __init__(self, api_key: str, *args: Any, **kwargs: Any):
        """Construct a HumeStreamClient.

        Args:
            api_key (str): Hume API key.
        """
        if not HAS_WEBSOCKETS:
            raise HumeClientException("The websockets package is required to use HumeStreamClient. "
                                      "Run `pip install hume[stream]` to install a version compatible with the"
                                      "Hume Python SDK.")

        super().__init__(api_key, *args, **kwargs)

    @classmethod
    def get_api_type(cls) -> ApiType:
        """Get the ApiType of the client.

        Returns:
            ApiType: API type of the client.
        """
        return ApiType.STREAM

    @asynccontextmanager
    async def connect(
        self,
        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.

        Args:
            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.
        """
        endpoint = self._construct_endpoint("models")
        try:
            # pylint: disable=no-member
            async with websockets.connect(  # type: ignore[attr-defined]
                    endpoint, extra_headers=self._get_client_headers()) as protocol:
                yield StreamSocket(protocol, configs, stream_window_ms=stream_window_ms)
        except websockets.exceptions.InvalidStatusCode as exc:
            status_code: int = exc.status_code
            if status_code == 401:  # Unauthorized
                message = "HumeStreamClient initialized with invalid API key."
                raise HumeClientException(message) from exc
            raise HumeClientException("Unexpected error when creating streaming connection") from exc

    @asynccontextmanager
    async def _connect_with_configs_dict(self, configs_dict: Any) -> AsyncIterator[StreamSocket]:
        """Connect to the streaming API with a single models configuration dict.

        Args:
            configs_dict (Any): Models configurations dict. This should be a dict from model name
                to model configuration dict. An empty dict uses the default configuration.
        """
        configs = deserialize_configs(configs_dict)
        async with self.connect(configs) as websocket:
            yield websocket

__init__(api_key, *args, **kwargs)

Construct a HumeStreamClient.

Parameters:

Name Type Description Default
api_key str

Hume API key.

 required
Source code in hume/_stream/hume_stream_client.py 
42
43
44
45
46
47
48
49
50
51
52
53
def __init__(self, api_key: str, *args: Any, **kwargs: Any):
    """Construct a HumeStreamClient.

    Args:
        api_key (str): Hume API key.
    """
    if not HAS_WEBSOCKETS:
        raise HumeClientException("The websockets package is required to use HumeStreamClient. "
                                  "Run `pip install hume[stream]` to install a version compatible with the"
                                  "Hume Python SDK.")

    super().__init__(api_key, *args, **kwargs)

connect(configs, stream_window_ms=None) async

Connect to the streaming API.

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.

Parameters:

Name Type Description Default
configs List[ModelConfigBase]

List of job configs.

 required
stream_window_ms Optional[int]

Length of the sliding window in milliseconds to use when aggregating media across streaming payloads within one websocket connection.

 None
Source code in hume/_stream/hume_stream_client.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@asynccontextmanager
async def connect(
    self,
    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.

    Args:
        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.
    """
    endpoint = self._construct_endpoint("models")
    try:
        # pylint: disable=no-member
        async with websockets.connect(  # type: ignore[attr-defined]
                endpoint, extra_headers=self._get_client_headers()) as protocol:
            yield StreamSocket(protocol, configs, stream_window_ms=stream_window_ms)
    except websockets.exceptions.InvalidStatusCode as exc:
        status_code: int = exc.status_code
        if status_code == 401:  # Unauthorized
            message = "HumeStreamClient initialized with invalid API key."
            raise HumeClientException(message) from exc
        raise HumeClientException("Unexpected error when creating streaming connection") from exc

get_api_type() classmethod

Get the ApiType of the client.

Returns:

Name Type Description
ApiType ApiType

API type of the client.
Source code in hume/_stream/hume_stream_client.py 
55
56
57
58
59
60
61
62
@classmethod
def get_api_type(cls) -> ApiType:
    """Get the ApiType of the client.

    Returns:
        ApiType: API type of the client.
    """
    return ApiType.STREAM