Source code for chariot.evaluation.model_comparison

from typing import Any

from chariot import _apis
from chariot_api._openapi.evaluation import models
from chariot_api._openapi.evaluation.models import (
    GPSBox as _GPSBox,
)
from chariot_api._openapi.evaluation.models import (
    MetadataFilter as _MetadataFilter,
)
from chariot_api._openapi.evaluation.models import (
    TimeRange as _TimeRange,
)




class GPSBox(_GPSBox):
    """GPS box coordinates.

    Attributes
    ----------
    min_lat : float | int
        Minimum latitude.
    min_lon : float | int
        Minimum longitude.
    max_lat : float | int
        Maximum latitude.
    max_lon : float | int
        Maximum longitude.

    """

    ...





class TimeRange(_TimeRange):
    """ISO-8601 format time boundaries.

    Attributes
    ----------
    start : str, optional
        Optional start time in ISO-8601 format.
    end : str, optional
        Optional end time in ISO-8601 format.

    """

    ...





class MetadataFilter(_MetadataFilter):
    """ISO-8601 format time boundaries.

    Attributes
    ----------
    key : str
        Metadata key to perform filtering upon.
    operator : str
        Operator symbol.
    value : str
        The value to compare by.

    """

    ...





@_apis.login_required
def evaluate(
    job_ids: list[str],
    filter_by_location: GPSBox | None = None,
    filter_by_time_range: TimeRange | None = None,
    filter_by_metadata: list[MetadataFilter] | None = None,
    aggregate_over_labels: list[str] | None = None,
) -> dict:
    """Run evaluation with option to filter data.

    Parameters
    ----------
    job_ids : list[str]
        A list of bulk inference job ids to perform evaluation over.
    filter_by_location : filter_by_location, optional
        Optionally filter datums by location.
    filter_by_time_range : TimeRange, optional
        Optionally filter datums by time of creation.
    filter_by_metadata : list[MetadataFilter], optional
        Optionally filter datums by metadata.
    aggregate_over_labels : list[str], optional
        Optionally create metrics over labels subsets.

    Returns
    -------
    dict
        The evaluation response.

    """
    filter_ = (
        models.EvaluationFilter(
            location=filter_by_location,
            time_range=filter_by_time_range,
            metadata=filter_by_metadata,
        )
        if filter_by_location or filter_by_time_range or filter_by_metadata
        else None
    )
    params = models.EvaluationParams(
        job_ids=job_ids,
        filter=filter_,
        aggregate_over_labels=aggregate_over_labels,
    )
    resp = _apis.evaluation.evaluation_api.filter_evaluation_evaluation_v1_evaluate_post(params)
    return resp.get("data")





@_apis.login_required
def get_metadata_keys(job_ids: list[str]) -> dict[str, Any]:
    """Get dataset metadata keys.

    Parameters
    ----------
    job_ids : list[str]
        A list of bulk inference job ids.

    """
    resp = _apis.evaluation.evaluation_api.get_metadata_keys_evaluation_v1_metadata_keys_get(
        job_ids
    )
    return resp.get("data")





@_apis.login_required
def get_metadata(job_ids: list[str]) -> list[dict[str, Any]]:
    """Get dataset metadata.

    Parameters
    ----------
    job_ids : list[str]
        A list of bulk inference job ids.

    """
    resp = _apis.evaluation.evaluation_api.get_metadata_evaluation_v1_metadata_get(job_ids=job_ids)
    return resp.get("data")