HumeBatchClient

Bases: ClientBase

Batch API client.

Example

from hume import HumeBatchClient

client = HumeBatchClient("<your-api-key>")
job = client.submit_face(["<your-image-url>"])

print(job)
print("Running...")

result = job.await_complete()
result.download_predictions("predictions.json")

print("Predictions downloaded!")

Source code in hume/_batch/hume_batch_client.py

class HumeBatchClient(ClientBase):
    """Batch API client.

    Example:
        ```python
        from hume import HumeBatchClient

        client = HumeBatchClient("<your-api-key>")
        job = client.submit_face(["<your-image-url>"])

        print(job)
        print("Running...")

        result = job.await_complete()
        result.download_predictions("predictions.json")

        print("Predictions downloaded!")
        ```
    """

    _DEFAULT_API_TIMEOUT = 10

    def __init__(self, *args: Any, **kwargs: Any):
        """Construct a HumeBatchClient.

        Args:
            api_key (str): Hume API key.
        """
        super().__init__(*args, **kwargs)

    def get_job_result(self, job_id: str) -> BatchJobResult:
        """Get the result of the batch job.

        Args:
            job_id (str): Job ID.

        Raises:
            HumeClientError: If the job result cannot be loaded.

        Returns:
            BatchJobResult: Batch job result.
        """
        endpoint = (f"{self._api_http_base_url}/{self._api_version}/{ApiType.BATCH.value}/jobs/{job_id}"
                    f"?apikey={self._api_key}")
        response = requests.get(endpoint, timeout=self._DEFAULT_API_TIMEOUT)
        try:
            body = response.json()
        except json.JSONDecodeError:
            # pylint: disable=raise-missing-from
            raise HumeClientError("Unexpected error when getting job result")

        if "message" in body and body["message"] == "job not found":
            raise HumeClientError(f"Could not find a job with ID {job_id}")

        return BatchJobResult.from_response(body)

    def submit_face(
        self,
        urls: List[str],
        fps_pred: Optional[float] = None,
        prob_threshold: Optional[float] = None,
        identify_faces: Optional[bool] = None,
        min_face_size: Optional[float] = None,
    ) -> BatchJob:
        """Submit a new job for facial expression.

        Args:
            urls (List[str]): URLs to process.
            fps_pred (Optional[float]): Number of frames per second to process. Other frames will be omitted
                from the response.
            prob_threshold (Optional[float]): Face detection probability threshold. Faces detected with a
                probability less than this threshold will be omitted from the response.
            identify_faces (Optional[bool]): Whether to return identifiers for faces across frames.
                If true, unique identifiers will be assigned to face bounding boxes to differentiate different faces.
                If false, all faces will be tagged with an "unknown" ID.
            min_face_size (Optional[float]): Minimum bounding box side length in pixels to treat as a face.
                Faces detected with a bounding box side length in pixels less than this threshold will be
                omitted from the response.

        Raises:
            HumeClientError: If the job fails.

        Returns:
            BatchJob: Batch job.
        """
        config = FaceConfig(
            fps_pred=fps_pred,
            prob_threshold=prob_threshold,
            identify_faces=identify_faces,
            min_face_size=min_face_size,
        )
        return self._submit(urls, [config])

    def submit_burst(
        self,
        urls: List[str],
    ) -> BatchJob:
        """Submit a new job for vocal bursts.

        Args:
            urls (List[str]): URLs to process.

        Raises:
            HumeClientError: If the job fails.

        Returns:
            BatchJob: Batch job.
        """
        config = BurstConfig()
        return self._submit(urls, [config])

    def submit_prosody(
        self,
        urls: List[str],
        identify_speakers: Optional[bool] = None,
    ) -> BatchJob:
        """Submit a new job for vocal bursts.

        Args:
            urls (List[str]): URLs to process.
            identify_speakers (Optional[bool]): Whether to return identifiers for speakers over time. If true,
                unique identifiers will be assigned to spoken words to differentiate different speakers. If false,
                all speakers will be tagged with an "unknown" ID.

        Raises:
            HumeClientError: If the job fails.

        Returns:
            BatchJob: Batch job.
        """
        config = ProsodyConfig(identify_speakers=identify_speakers)
        return self._submit(urls, [config])

    def submit_language(
        self,
        urls: List[str],
        sliding_window: Optional[bool] = None,
        identify_speakers: Optional[bool] = None,
    ) -> BatchJob:
        """Submit a new job for language emotion.

        Args:
            urls (List[str]): URLs to process.
            sliding_window (Optional[float]): Whether to generate predictions for each word in the text or
                for the entire text in aggregate.
            identify_speakers (Optional[bool]): Whether to return identifiers for speakers over time.
                If true, unique identifiers will be assigned to spoken words to differentiate different speakers.
                If false, all speakers will be tagged with an "unknown" ID.

        Raises:
            HumeClientError: If the job fails.

        Returns:
            BatchJob: Batch job.
        """
        config = LanguageConfig(
            sliding_window=sliding_window,
            identify_speakers=identify_speakers,
        )
        return self._submit(urls, [config])

    def _submit(self, urls: List[str], configs: List[JobConfigBase]) -> BatchJob:
        request = self._get_request(configs, urls)
        return self.start_job(request)

    def get_job(self, job_id: str) -> BatchJob:
        """Rehydrate a job based on a Job ID.

        Args:
            job_id (str): ID of the job to rehydrate.

        Returns:
            BatchJob: Job associated with the given ID.
        """
        return BatchJob(self, job_id)

    def start_job(self, request_body: Any) -> BatchJob:
        """Start a batch job.

        Args:
            request_body (Any): JSON request body to be passed to the batch API.

        Raises:
            HumeClientError: If the batch job fails to start.

        Returns:
            BatchJob: A `BatchJob` that wraps the batch computation.
        """
        endpoint = (f"{self._api_http_base_url}/{self._api_version}/{ApiType.BATCH.value}/jobs"
                    f"?apikey={self._api_key}")
        response = requests.post(endpoint, json=request_body, timeout=self._DEFAULT_API_TIMEOUT)

        try:
            body = response.json()
        except json.decoder.JSONDecodeError:
            # pylint: disable=raise-missing-from
            raise HumeClientError(f"Failed batch request: {response.text}")

        if "job_id" not in body:
            if "fault" in body and "faultstring" in body["fault"]:
                fault_string = body["fault"]["faultstring"]
                raise HumeClientError(f"Could not start batch job: {fault_string}")
            raise HumeClientError("Unexpected error when starting batch job")

        return BatchJob(self, body["job_id"])

    @classmethod
    def _get_request(cls, configs: List[JobConfigBase], urls: List[str]) -> Dict[str, Any]:
        model_requests = {}
        for config in configs:
            model_requests[config.get_model_type().value] = config.serialize()

        return {
            "models": model_requests,
            "urls": urls,
        }

__init__(*args, **kwargs)

Construct a HumeBatchClient.

Parameters:

Name	Type	Description	Default
api_key	str	Hume API key.	required

Source code in hume/_batch/hume_batch_client.py

def __init__(self, *args: Any, **kwargs: Any):
    """Construct a HumeBatchClient.

    Args:
        api_key (str): Hume API key.
    """
    super().__init__(*args, **kwargs)

get_job(job_id)

Rehydrate a job based on a Job ID.

Parameters:

Name	Type	Description	Default
job_id	str	ID of the job to rehydrate.	required

Returns:

Name	Type	Description
BatchJob	BatchJob	Job associated with the given ID.

Source code in hume/_batch/hume_batch_client.py

def get_job(self, job_id: str) -> BatchJob:
    """Rehydrate a job based on a Job ID.

    Args:
        job_id (str): ID of the job to rehydrate.

    Returns:
        BatchJob: Job associated with the given ID.
    """
    return BatchJob(self, job_id)

get_job_result(job_id)

Get the result of the batch job.

Parameters:

Name	Type	Description	Default
job_id	str	Job ID.	required

Raises:

Type	Description
HumeClientError	If the job result cannot be loaded.

Returns:

Name	Type	Description
BatchJobResult	BatchJobResult	Batch job result.

Source code in hume/_batch/hume_batch_client.py

def get_job_result(self, job_id: str) -> BatchJobResult:
    """Get the result of the batch job.

    Args:
        job_id (str): Job ID.

    Raises:
        HumeClientError: If the job result cannot be loaded.

    Returns:
        BatchJobResult: Batch job result.
    """
    endpoint = (f"{self._api_http_base_url}/{self._api_version}/{ApiType.BATCH.value}/jobs/{job_id}"
                f"?apikey={self._api_key}")
    response = requests.get(endpoint, timeout=self._DEFAULT_API_TIMEOUT)
    try:
        body = response.json()
    except json.JSONDecodeError:
        # pylint: disable=raise-missing-from
        raise HumeClientError("Unexpected error when getting job result")

    if "message" in body and body["message"] == "job not found":
        raise HumeClientError(f"Could not find a job with ID {job_id}")

    return BatchJobResult.from_response(body)

start_job(request_body)

Start a batch job.

Parameters:

Name	Type	Description	Default
request_body	Any	JSON request body to be passed to the batch API.	required

Raises:

Type	Description
HumeClientError	If the batch job fails to start.

Returns:

Name	Type	Description
BatchJob	BatchJob	A BatchJob that wraps the batch computation.

Source code in hume/_batch/hume_batch_client.py

def start_job(self, request_body: Any) -> BatchJob:
    """Start a batch job.

    Args:
        request_body (Any): JSON request body to be passed to the batch API.

    Raises:
        HumeClientError: If the batch job fails to start.

    Returns:
        BatchJob: A `BatchJob` that wraps the batch computation.
    """
    endpoint = (f"{self._api_http_base_url}/{self._api_version}/{ApiType.BATCH.value}/jobs"
                f"?apikey={self._api_key}")
    response = requests.post(endpoint, json=request_body, timeout=self._DEFAULT_API_TIMEOUT)

    try:
        body = response.json()
    except json.decoder.JSONDecodeError:
        # pylint: disable=raise-missing-from
        raise HumeClientError(f"Failed batch request: {response.text}")

    if "job_id" not in body:
        if "fault" in body and "faultstring" in body["fault"]:
            fault_string = body["fault"]["faultstring"]
            raise HumeClientError(f"Could not start batch job: {fault_string}")
        raise HumeClientError("Unexpected error when starting batch job")

    return BatchJob(self, body["job_id"])

submit_burst(urls)

Submit a new job for vocal bursts.

Parameters:

Name	Type	Description	Default
urls	List[str]	URLs to process.	required

Raises:

Type	Description
HumeClientError	If the job fails.

Returns:

Name	Type	Description
BatchJob	BatchJob	Batch job.

Source code in hume/_batch/hume_batch_client.py

def submit_burst(
    self,
    urls: List[str],
) -> BatchJob:
    """Submit a new job for vocal bursts.

    Args:
        urls (List[str]): URLs to process.

    Raises:
        HumeClientError: If the job fails.

    Returns:
        BatchJob: Batch job.
    """
    config = BurstConfig()
    return self._submit(urls, [config])

submit_face(urls, fps_pred=None, prob_threshold=None, identify_faces=None, min_face_size=None)

Submit a new job for facial expression.

Parameters:

Name	Type	Description	Default
urls	List[str]	URLs to process.	required
fps_pred	Optional[float]	Number of frames per second to process. Other frames will be omitted from the response.	None
prob_threshold	Optional[float]	Face detection probability threshold. Faces detected with a probability less than this threshold will be omitted from the response.	None
identify_faces	Optional[bool]	Whether to return identifiers for faces across frames. If true, unique identifiers will be assigned to face bounding boxes to differentiate different faces. If false, all faces will be tagged with an "unknown" ID.	None
min_face_size	Optional[float]	Minimum bounding box side length in pixels to treat as a face. Faces detected with a bounding box side length in pixels less than this threshold will be omitted from the response.	None

Raises:

Type	Description
HumeClientError	If the job fails.

Returns:

Name	Type	Description
BatchJob	BatchJob	Batch job.

Source code in hume/_batch/hume_batch_client.py

def submit_face(
    self,
    urls: List[str],
    fps_pred: Optional[float] = None,
    prob_threshold: Optional[float] = None,
    identify_faces: Optional[bool] = None,
    min_face_size: Optional[float] = None,
) -> BatchJob:
    """Submit a new job for facial expression.

    Args:
        urls (List[str]): URLs to process.
        fps_pred (Optional[float]): Number of frames per second to process. Other frames will be omitted
            from the response.
        prob_threshold (Optional[float]): Face detection probability threshold. Faces detected with a
            probability less than this threshold will be omitted from the response.
        identify_faces (Optional[bool]): Whether to return identifiers for faces across frames.
            If true, unique identifiers will be assigned to face bounding boxes to differentiate different faces.
            If false, all faces will be tagged with an "unknown" ID.
        min_face_size (Optional[float]): Minimum bounding box side length in pixels to treat as a face.
            Faces detected with a bounding box side length in pixels less than this threshold will be
            omitted from the response.

    Raises:
        HumeClientError: If the job fails.

    Returns:
        BatchJob: Batch job.
    """
    config = FaceConfig(
        fps_pred=fps_pred,
        prob_threshold=prob_threshold,
        identify_faces=identify_faces,
        min_face_size=min_face_size,
    )
    return self._submit(urls, [config])

submit_language(urls, sliding_window=None, identify_speakers=None)

Submit a new job for language emotion.

Parameters:

Name	Type	Description	Default
urls	List[str]	URLs to process.	required
sliding_window	Optional[float]	Whether to generate predictions for each word in the text or for the entire text in aggregate.	None
identify_speakers	Optional[bool]	Whether to return identifiers for speakers over time. If true, unique identifiers will be assigned to spoken words to differentiate different speakers. If false, all speakers will be tagged with an "unknown" ID.	None

Raises:

Type	Description
HumeClientError	If the job fails.

Returns:

Name	Type	Description
BatchJob	BatchJob	Batch job.

Source code in hume/_batch/hume_batch_client.py

def submit_language(
    self,
    urls: List[str],
    sliding_window: Optional[bool] = None,
    identify_speakers: Optional[bool] = None,
) -> BatchJob:
    """Submit a new job for language emotion.

    Args:
        urls (List[str]): URLs to process.
        sliding_window (Optional[float]): Whether to generate predictions for each word in the text or
            for the entire text in aggregate.
        identify_speakers (Optional[bool]): Whether to return identifiers for speakers over time.
            If true, unique identifiers will be assigned to spoken words to differentiate different speakers.
            If false, all speakers will be tagged with an "unknown" ID.

    Raises:
        HumeClientError: If the job fails.

    Returns:
        BatchJob: Batch job.
    """
    config = LanguageConfig(
        sliding_window=sliding_window,
        identify_speakers=identify_speakers,
    )
    return self._submit(urls, [config])

submit_prosody(urls, identify_speakers=None)

Submit a new job for vocal bursts.

Parameters:

Name	Type	Description	Default
urls	List[str]	URLs to process.	required
identify_speakers	Optional[bool]	Whether to return identifiers for speakers over time. If true, unique identifiers will be assigned to spoken words to differentiate different speakers. If false, all speakers will be tagged with an "unknown" ID.	None

Raises:

Type	Description
HumeClientError	If the job fails.

Returns:

Name	Type	Description
BatchJob	BatchJob	Batch job.

Source code in hume/_batch/hume_batch_client.py

def submit_prosody(
    self,
    urls: List[str],
    identify_speakers: Optional[bool] = None,
) -> BatchJob:
    """Submit a new job for vocal bursts.

    Args:
        urls (List[str]): URLs to process.
        identify_speakers (Optional[bool]): Whether to return identifiers for speakers over time. If true,
            unique identifiers will be assigned to spoken words to differentiate different speakers. If false,
            all speakers will be tagged with an "unknown" ID.

    Raises:
        HumeClientError: If the job fails.

    Returns:
        BatchJob: Batch job.
    """
    config = ProsodyConfig(identify_speakers=identify_speakers)
    return self._submit(urls, [config])