Skip to content

explanation_computing

Explanation computing job utilities.

Functions:

Name Description
get_job_result

Get explanation job results.

get_job_result(job_id: UUID) -> ExplanationArtifact #

Get explanation job results.

Source code in src/xpdeep/utils/jobs/explanation_computing.py
@initialized_client_verification
@initialized_project_verification
def get_job_result(job_id: UUID) -> ExplanationArtifact:
    """Get explanation job results."""
    try:
        logging.print_job_progress(job_id)
    except KeyboardInterrupt:
        handle_api_validation_errors(
            cancel_job.sync(Project.CURRENT.get().model.id, job_id, client=ClientFactory.CURRENT.get()())
        )
        raise

    ongoing_job = handle_api_validation_errors(
        get_one_job.sync(Project.CURRENT.get().model.id, job_id, client=ClientFactory.CURRENT.get()()),
    )

    if ongoing_job.status == JobStatus.ERROR:
        remote_error = cast(JobModelResultsType0, ongoing_job.results)
        error_message = f"{remote_error['error_class']} : {remote_error['error_detail']}"
        raise ApiError(error_message)

    if ongoing_job.results is None:
        message = "Job failed"
        raise ApiError(message)

    if ongoing_job.type_ not in {JobType.GLOBAL_EXPLAIN, JobType.INFERENCE}:
        message = f"Unexpected Job type {ongoing_job.type_}."
        raise ApiError(message)

    return ExplanationArtifact(
        handle_api_validation_errors(
            get_one_explanation.sync(
                Project.CURRENT.get().model.id,
                ongoing_job.results["id"],
                client=ClientFactory.CURRENT.get()(),
            ),
        )
    )