Skip to content

StreamSocket

Streaming socket connection.

Source code in hume/_stream/stream_socket.py 
 18
 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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
class StreamSocket:
    """Streaming socket connection."""

    _FACE_LIMIT = 100
    _N_LANDMARKS = 478
    _N_SPATIAL = 3

    def __init__(
        self,
        protocol: "WebSocketClientProtocol",
        configs: List[ModelConfigBase],
        stream_window_ms: Optional[int] = None,
    ):
        """Construct a `StreamSocket`.

        Args:
            protocol (WebSocketClientProtocol): Protocol instance from websockets library.
            configs (Optional[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.
        """
        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.")

        self._protocol = protocol
        self._configs = configs
        self._stream_window_ms = stream_window_ms

        # Serialize configs once for full lifetime of socket
        self._serialized_configs = serialize_configs(configs)

    async def send_file(
        self,
        filepath: Union[str, Path],
        configs: Optional[List[ModelConfigBase]] = None,
    ) -> Any:
        """Send a file on the `StreamSocket`.

        Args:
            filepath (Path): Path to media file to send on socket connection.
            configs (Optional[List[ModelConfigBase]]): List of model configurations.
                If set these configurations will overwrite any configurations
                set when initializing the `StreamSocket`.

        Returns:
            Any: Response from the streaming API.
        """
        with Path(filepath).open('rb') as f:
            bytes_data = base64.b64encode(f.read())
            return await self.send_bytes(bytes_data, configs=configs)

    async def send_bytes(
        self,
        bytes_data: bytes,
        configs: Optional[List[ModelConfigBase]] = None,
    ) -> Any:
        """Send raw bytes on the `StreamSocket`.

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

        Args:
            bytes_data (bytes): Raw bytes of media to send on socket connection.
            configs (Optional[List[ModelConfigBase]]): List of model configurations.
                If set these configurations will overwrite any configurations
                set when initializing the `StreamSocket`.

        Returns:
            Any: Response from the streaming API.
        """
        bytes_str = bytes_data.decode("utf-8")
        return await self._send_str(bytes_str, raw_text=False, configs=configs)

    async def send_text(
        self,
        text: str,
        configs: Optional[List[ModelConfigBase]] = None,
    ) -> Any:
        """Send text on the `StreamSocket`.

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

        Args:
            text (str): Text to send to the language model.
            configs (Optional[List[ModelConfigBase]]): List of model configurations.
                If set these configurations will overwrite any configurations
                set when initializing the `StreamSocket`.

        Raises:
            HumeClientException: If the socket is configured with a modality other than language.

        Returns:
            Any: Response from the streaming API.
        """
        self._validate_configs_with_model_type(LanguageConfig, "send_text", configs=configs)
        return await self._send_str(text, raw_text=True, configs=configs)

    async def send_facemesh(
        self,
        landmarks: List[List[List[float]]],
        configs: Optional[List[ModelConfigBase]] = None,
    ) -> Any:
        """Send facemesh landmarks on the `StreamSocket`.

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

        Args:
            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.
            configs (Optional[List[ModelConfigBase]]): List of model configurations.
                If set these configurations will overwrite any configurations
                set when initializing the `StreamSocket`.

        Raises:
            HumeClientException: If the socket is configured with a modality other than facemesh.

        Returns:
            Any: Response from the streaming API.
        """
        self._validate_configs_with_model_type(FacemeshConfig, "send_facemesh", configs=configs)

        n_faces = len(landmarks)
        if n_faces > self._FACE_LIMIT:
            raise HumeClientException("Number of faces sent in facemesh payload was greater "
                                      f"than the limit of {self._FACE_LIMIT}, found {n_faces}.")
        if n_faces == 0:
            raise HumeClientException("No faces sent in facemesh payload.")
        n_landmarks = len(landmarks[0])
        if n_landmarks != self._N_LANDMARKS:
            raise HumeClientException(f"Number of MediaPipe landmarks per face must be exactly {self._N_LANDMARKS}, "
                                      f"found {n_landmarks}.")
        if len(landmarks[0][0]) != self._N_SPATIAL:
            raise HumeClientException("Invalid facemesh payload detected. "
                                      "Each facemesh landmark should be an (x, y, z) point.")

        landmarks_str = json.dumps(landmarks)
        bytes_data = base64.b64encode(landmarks_str.encode("utf-8"))
        return await self.send_bytes(bytes_data, configs=configs)

    async def reset_stream(self) -> 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.
        """
        payload = {
            "reset_stream": True,
        }
        return await self._send_payload(payload)

    async def get_job_details(self) -> Any:
        """Get details associated with the current streaming connection.

        Returns:
            Any: Response from the streaming API.
        """
        payload = {
            "job_details": True,
        }
        return await self._send_payload(payload)

    async def _send_str(
        self,
        data: str,
        *,
        raw_text: bool,
        configs: Optional[List[ModelConfigBase]] = None,
    ) -> Any:
        serialized_configs = self._serialized_configs
        if configs is not None:
            serialized_configs = serialize_configs(configs)

        payload: Dict[str, Any] = {
            "data": data,
            "models": serialized_configs,
            "raw_text": raw_text,
        }
        if self._stream_window_ms is not None:
            payload["stream_window_ms"] = self._stream_window_ms
        return await self._send_payload(payload)

    async def _send_payload(self, payload: Dict[str, Any]) -> Any:
        request_message = json.dumps(payload)
        await self._protocol.send(request_message)
        response_data = await self._protocol.recv()
        # Cast to str because websockets can send bytes, but we will always accept JSON strings
        response_str = str(response_data)

        try:
            response = json.loads(response_str)
        except json.JSONDecodeError as exc:
            raise HumeClientException("Unexpected error when fetching streaming API predictions") from exc

        if "error" in response:
            error = response["error"]
            code = response["code"]
            raise HumeClientException.from_error(code, error)

        return response

    def _validate_configs_with_model_type(
        self,
        config_type: Any,
        method_name: str,
        configs: Optional[List[ModelConfigBase]] = None,
    ) -> None:
        config_method = "Socket"
        payload_configs = self._configs
        if configs is not None:
            config_method = "Payload"
            payload_configs = configs

        for config in payload_configs:
            if not isinstance(config, config_type):
                config_name = config_type.__name__
                invalid_config_name = config.__class__.__name__
                raise HumeClientException(f"{config_method} configured with {invalid_config_name}. "
                                          f"{method_name} is only supported when using a {config_name}.")

__init__(protocol, configs, stream_window_ms=None)

Construct a StreamSocket.

Parameters:

Name Type Description Default
protocol WebSocketClientProtocol

Protocol instance from websockets library.

 required
configs Optional[List[ModelConfigBase]]

List of model configurations.

 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

Raises:

Type Description
HumeClientException

If there is an error processing media over the socket connection.
Source code in hume/_stream/stream_socket.py 
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
def __init__(
    self,
    protocol: "WebSocketClientProtocol",
    configs: List[ModelConfigBase],
    stream_window_ms: Optional[int] = None,
):
    """Construct a `StreamSocket`.

    Args:
        protocol (WebSocketClientProtocol): Protocol instance from websockets library.
        configs (Optional[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.
    """
    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.")

    self._protocol = protocol
    self._configs = configs
    self._stream_window_ms = stream_window_ms

    # Serialize configs once for full lifetime of socket
    self._serialized_configs = serialize_configs(configs)

get_job_details() async

Get details associated with the current streaming connection.

Returns:

Name Type Description
Any Any

Response from the streaming API.
Source code in hume/_stream/stream_socket.py 
181
182
183
184
185
186
187
188
189
190
async def get_job_details(self) -> Any:
    """Get details associated with the current streaming connection.

    Returns:
        Any: Response from the streaming API.
    """
    payload = {
        "job_details": True,
    }
    return await self._send_payload(payload)

reset_stream() async

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:

Name Type Description
Any Any

Response from the streaming API.
Source code in hume/_stream/stream_socket.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
async def reset_stream(self) -> 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.
    """
    payload = {
        "reset_stream": True,
    }
    return await self._send_payload(payload)

send_bytes(bytes_data, configs=None) async

Send raw bytes on the StreamSocket.

Input should be base64 encoded bytes.

You can use base64.b64encode() to encode a raw string.

Parameters:

Name Type Description Default
bytes_data bytes

Raw bytes of media to send on socket connection.

 required
configs Optional[List[ModelConfigBase]]

List of model configurations. If set these configurations will overwrite any configurations set when initializing the StreamSocket.

 None

Returns:

Name Type Description
Any Any

Response from the streaming API.
Source code in hume/_stream/stream_socket.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
async def send_bytes(
    self,
    bytes_data: bytes,
    configs: Optional[List[ModelConfigBase]] = None,
) -> Any:
    """Send raw bytes on the `StreamSocket`.

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

    Args:
        bytes_data (bytes): Raw bytes of media to send on socket connection.
        configs (Optional[List[ModelConfigBase]]): List of model configurations.
            If set these configurations will overwrite any configurations
            set when initializing the `StreamSocket`.

    Returns:
        Any: Response from the streaming API.
    """
    bytes_str = bytes_data.decode("utf-8")
    return await self._send_str(bytes_str, raw_text=False, configs=configs)

send_facemesh(landmarks, configs=None) async

Send facemesh landmarks on the StreamSocket.

This method is intended for use with a FacemeshConfig.

When the socket is configured for other modalities this method will fail.

Parameters:

Name Type Description Default
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.

 required
configs Optional[List[ModelConfigBase]]

List of model configurations. If set these configurations will overwrite any configurations set when initializing the StreamSocket.

 None

Raises:

Type Description
HumeClientException

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

Returns:

Name Type Description
Any Any

Response from the streaming API.
Source code in hume/_stream/stream_socket.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
async def send_facemesh(
    self,
    landmarks: List[List[List[float]]],
    configs: Optional[List[ModelConfigBase]] = None,
) -> Any:
    """Send facemesh landmarks on the `StreamSocket`.

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

    Args:
        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.
        configs (Optional[List[ModelConfigBase]]): List of model configurations.
            If set these configurations will overwrite any configurations
            set when initializing the `StreamSocket`.

    Raises:
        HumeClientException: If the socket is configured with a modality other than facemesh.

    Returns:
        Any: Response from the streaming API.
    """
    self._validate_configs_with_model_type(FacemeshConfig, "send_facemesh", configs=configs)

    n_faces = len(landmarks)
    if n_faces > self._FACE_LIMIT:
        raise HumeClientException("Number of faces sent in facemesh payload was greater "
                                  f"than the limit of {self._FACE_LIMIT}, found {n_faces}.")
    if n_faces == 0:
        raise HumeClientException("No faces sent in facemesh payload.")
    n_landmarks = len(landmarks[0])
    if n_landmarks != self._N_LANDMARKS:
        raise HumeClientException(f"Number of MediaPipe landmarks per face must be exactly {self._N_LANDMARKS}, "
                                  f"found {n_landmarks}.")
    if len(landmarks[0][0]) != self._N_SPATIAL:
        raise HumeClientException("Invalid facemesh payload detected. "
                                  "Each facemesh landmark should be an (x, y, z) point.")

    landmarks_str = json.dumps(landmarks)
    bytes_data = base64.b64encode(landmarks_str.encode("utf-8"))
    return await self.send_bytes(bytes_data, configs=configs)

send_file(filepath, configs=None) async

Send a file on the StreamSocket.

Parameters:

Name Type Description Default
filepath Path

Path to media file to send on socket connection.

 required
configs Optional[List[ModelConfigBase]]

List of model configurations. If set these configurations will overwrite any configurations set when initializing the StreamSocket.

 None

Returns:

Name Type Description
Any Any

Response from the streaming API.
Source code in hume/_stream/stream_socket.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
async def send_file(
    self,
    filepath: Union[str, Path],
    configs: Optional[List[ModelConfigBase]] = None,
) -> Any:
    """Send a file on the `StreamSocket`.

    Args:
        filepath (Path): Path to media file to send on socket connection.
        configs (Optional[List[ModelConfigBase]]): List of model configurations.
            If set these configurations will overwrite any configurations
            set when initializing the `StreamSocket`.

    Returns:
        Any: Response from the streaming API.
    """
    with Path(filepath).open('rb') as f:
        bytes_data = base64.b64encode(f.read())
        return await self.send_bytes(bytes_data, configs=configs)

send_text(text, configs=None) async

Send text on the StreamSocket.

This method is intended for use with a LanguageConfig.

When the socket is configured for other modalities this method will fail.

Parameters:

Name Type Description Default
text str

Text to send to the language model.

 required
configs Optional[List[ModelConfigBase]]

List of model configurations. If set these configurations will overwrite any configurations set when initializing the StreamSocket.

 None

Raises:

Type Description
HumeClientException

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

Returns:

Name Type Description
Any Any

Response from the streaming API.
Source code in hume/_stream/stream_socket.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
async def send_text(
    self,
    text: str,
    configs: Optional[List[ModelConfigBase]] = None,
) -> Any:
    """Send text on the `StreamSocket`.

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

    Args:
        text (str): Text to send to the language model.
        configs (Optional[List[ModelConfigBase]]): List of model configurations.
            If set these configurations will overwrite any configurations
            set when initializing the `StreamSocket`.

    Raises:
        HumeClientException: If the socket is configured with a modality other than language.

    Returns:
        Any: Response from the streaming API.
    """
    self._validate_configs_with_model_type(LanguageConfig, "send_text", configs=configs)
    return await self._send_str(text, raw_text=True, configs=configs)