Documentation

Documentation

valor_lite.classification.Classification dataclass

Classification data structure containing a ground truth label and a list of predictions.

Parameters:

Name	Type	Description	Default
uid	str	Unique identifier for the instance.	required
groundtruth	str	The true label for the instance.	required
predictions	list of str	List of predicted labels.	required
scores	list of float	Confidence scores corresponding to each predicted label.	required

Examples:

>>> classification = Classification(
...     uid='123',
...     groundtruth='cat',
...     predictions=['cat', 'dog', 'bird'],
...     scores=[0.9, 0.05, 0.05]
... )

Source code in valor_lite/classification/annotation.py

@dataclass
class Classification:
    """
    Classification data structure containing a ground truth label and a list of predictions.

    Parameters
    ----------
    uid : str
        Unique identifier for the instance.
    groundtruth : str
        The true label for the instance.
    predictions : list of str
        List of predicted labels.
    scores : list of float
        Confidence scores corresponding to each predicted label.

    Examples
    --------
    >>> classification = Classification(
    ...     uid='123',
    ...     groundtruth='cat',
    ...     predictions=['cat', 'dog', 'bird'],
    ...     scores=[0.9, 0.05, 0.05]
    ... )
    """

    uid: str
    groundtruth: str
    predictions: list[str]
    scores: list[float]

    def __post_init__(self):
        if not isinstance(self.groundtruth, str):
            raise ValueError(
                "A classification must contain a single groundtruth."
            )
        if len(self.predictions) != len(self.scores):
            raise ValueError("There must be a score per prediction label.")

valor_lite.classification.DataLoader

Bases: Evaluator

Used for backwards compatibility as the Evaluator now handles ingestion.

Source code in valor_lite/classification/manager.py

class DataLoader(Evaluator):
    """
    Used for backwards compatibility as the Evaluator now handles ingestion.
    """

    pass

valor_lite.classification.Evaluator

Classification Evaluator

Source code in valor_lite/classification/manager.py

class Evaluator:
    """
    Classification Evaluator
    """

    def __init__(self):
        # external references
        self.datum_id_to_index: dict[str, int] = {}
        self.label_to_index: dict[str, int] = {}

        self.index_to_datum_id: list[str] = []
        self.index_to_label: list[str] = []

        # internal caches
        self._detailed_pairs = np.array([])
        self._label_metadata = np.array([], dtype=np.int32)
        self._metadata = Metadata()

    @property
    def metadata(self) -> Metadata:
        return self._metadata

    @property
    def ignored_prediction_labels(self) -> list[str]:
        """
        Prediction labels that are not present in the ground truth set.
        """
        glabels = set(np.where(self._label_metadata[:, 0] > 0)[0])
        plabels = set(np.where(self._label_metadata[:, 1] > 0)[0])
        return [
            self.index_to_label[label_id] for label_id in (plabels - glabels)
        ]

    @property
    def missing_prediction_labels(self) -> list[str]:
        """
        Ground truth labels that are not present in the prediction set.
        """
        glabels = set(np.where(self._label_metadata[:, 0] > 0)[0])
        plabels = set(np.where(self._label_metadata[:, 1] > 0)[0])
        return [
            self.index_to_label[label_id] for label_id in (glabels - plabels)
        ]

    def create_filter(
        self,
        datum_ids: list[str] | None = None,
        labels: list[str] | None = None,
    ) -> Filter:
        """
        Creates a filter object.

        Parameters
        ----------
        datum_uids : list[str], optional
            An optional list of string uids representing datums.
        labels : list[str], optional
            An optional list of labels.

        Returns
        -------
        Filter
            The filter object representing the input parameters.
        """
        # create datum mask
        n_pairs = self._detailed_pairs.shape[0]
        datum_mask = np.ones(n_pairs, dtype=np.bool_)
        if datum_ids is not None:
            if not datum_ids:
                return Filter(
                    datum_mask=np.zeros_like(datum_mask),
                    valid_label_indices=None,
                    metadata=Metadata(),
                )
            valid_datum_indices = np.array(
                [self.datum_id_to_index[uid] for uid in datum_ids],
                dtype=np.int32,
            )
            datum_mask = np.isin(
                self._detailed_pairs[:, 0], valid_datum_indices
            )

        # collect valid label indices
        valid_label_indices = None
        if labels is not None:
            if not labels:
                return Filter(
                    datum_mask=datum_mask,
                    valid_label_indices=np.array([], dtype=np.int32),
                    metadata=Metadata(),
                )
            valid_label_indices = np.array(
                [self.label_to_index[label] for label in labels] + [-1]
            )

        filtered_detailed_pairs, _ = filter_cache(
            detailed_pairs=self._detailed_pairs,
            datum_mask=datum_mask,
            valid_label_indices=valid_label_indices,
            n_labels=self.metadata.number_of_labels,
        )

        number_of_datums = (
            len(datum_ids)
            if datum_ids is not None
            else self.metadata.number_of_datums
        )

        return Filter(
            datum_mask=datum_mask,
            valid_label_indices=valid_label_indices,
            metadata=Metadata.create(
                detailed_pairs=filtered_detailed_pairs,
                number_of_datums=number_of_datums,
                number_of_labels=self.metadata.number_of_labels,
            ),
        )

    def filter(
        self, filter_: Filter
    ) -> tuple[NDArray[np.float64], NDArray[np.int32]]:
        """
        Performs filtering over the internal cache.

        Parameters
        ----------
        filter_ : Filter
            The filter object representation.

        Returns
        -------
        NDArray[float64]
            The filtered detailed pairs.
        NDArray[int32]
            The filtered label metadata.
        """
        return filter_cache(
            detailed_pairs=self._detailed_pairs,
            datum_mask=filter_.datum_mask,
            valid_label_indices=filter_.valid_label_indices,
            n_labels=self.metadata.number_of_labels,
        )

    def compute_precision_recall_rocauc(
        self,
        score_thresholds: list[float] = [0.0],
        hardmax: bool = True,
        filter_: Filter | None = None,
    ) -> dict[MetricType, list]:
        """
        Performs an evaluation and returns metrics.

        Parameters
        ----------
        score_thresholds : list[float]
            A list of score thresholds to compute metrics over.
        hardmax : bool
            Toggles whether a hardmax is applied to predictions.
        filter_ : Filter, optional
            Applies a filter to the internal cache.

        Returns
        -------
        dict[MetricType, list]
            A dictionary mapping MetricType enumerations to lists of computed metrics.
        """
        # apply filters
        if filter_ is not None:
            detailed_pairs, label_metadata = self.filter(filter_=filter_)
            n_datums = filter_.metadata.number_of_datums
        else:
            detailed_pairs = self._detailed_pairs
            label_metadata = self._label_metadata
            n_datums = self.metadata.number_of_datums

        results = compute_precision_recall_rocauc(
            detailed_pairs=detailed_pairs,
            label_metadata=label_metadata,
            score_thresholds=np.array(score_thresholds),
            hardmax=hardmax,
            n_datums=n_datums,
        )
        return unpack_precision_recall_rocauc_into_metric_lists(
            results=results,
            score_thresholds=score_thresholds,
            hardmax=hardmax,
            label_metadata=label_metadata,
            index_to_label=self.index_to_label,
        )

    def compute_confusion_matrix(
        self,
        score_thresholds: list[float] = [0.0],
        hardmax: bool = True,
        filter_: Filter | None = None,
    ) -> list[Metric]:
        """
        Computes a detailed confusion matrix..

        Parameters
        ----------
        score_thresholds : list[float]
            A list of score thresholds to compute metrics over.
        hardmax : bool
            Toggles whether a hardmax is applied to predictions.
        filter_ : Filter, optional
            Applies a filter to the internal cache.

        Returns
        -------
        list[Metric]
            A list of confusion matrices.
        """
        # apply filters
        if filter_ is not None:
            detailed_pairs, _ = self.filter(filter_=filter_)
        else:
            detailed_pairs = self._detailed_pairs

        if detailed_pairs.size == 0:
            return list()

        result = compute_confusion_matrix(
            detailed_pairs=detailed_pairs,
            score_thresholds=np.array(score_thresholds),
            hardmax=hardmax,
        )
        return unpack_confusion_matrix_into_metric_list(
            detailed_pairs=detailed_pairs,
            result=result,
            score_thresholds=score_thresholds,
            index_to_datum_id=self.index_to_datum_id,
            index_to_label=self.index_to_label,
        )

    def evaluate(
        self,
        score_thresholds: list[float] = [0.0],
        hardmax: bool = True,
        filter_: Filter | None = None,
    ) -> dict[MetricType, list[Metric]]:
        """
        Computes a detailed confusion matrix..

        Parameters
        ----------
        score_thresholds : list[float]
            A list of score thresholds to compute metrics over.
        hardmax : bool
            Toggles whether a hardmax is applied to predictions.
        filter_ : Filter, optional
            Applies a filter to the internal cache.

        Returns
        -------
        dict[MetricType, list[Metric]]
            Lists of metrics organized by metric type.
        """
        metrics = self.compute_precision_recall_rocauc(
            score_thresholds=score_thresholds,
            hardmax=hardmax,
            filter_=filter_,
        )
        metrics[MetricType.ConfusionMatrix] = self.compute_confusion_matrix(
            score_thresholds=score_thresholds,
            hardmax=hardmax,
            filter_=filter_,
        )
        return metrics

    def _add_datum(self, uid: str) -> int:
        """
        Helper function for adding a datum to the cache.

        Parameters
        ----------
        uid : str
            The datum uid.

        Returns
        -------
        int
            The datum index.

        Raises
        ------
        ValueError
            If datum id already exists.
        """
        if uid in self.datum_id_to_index:
            raise ValueError("datum with id '{uid}' already exists")
        index = len(self.datum_id_to_index)
        self.datum_id_to_index[uid] = index
        self.index_to_datum_id.append(uid)
        return self.datum_id_to_index[uid]

    def _add_label(self, label: str) -> int:
        """
        Helper function for adding a label to the cache.

        Parameters
        ----------
        label : str
            A string representing a label.

        Returns
        -------
        int
            Label index.
        """
        label_id = len(self.index_to_label)
        if label not in self.label_to_index:
            self.label_to_index[label] = label_id
            self.index_to_label.append(label)
            label_id += 1
        return self.label_to_index[label]

    def add_data(
        self,
        classifications: list[Classification],
        show_progress: bool = False,
    ):
        """
        Adds classifications to the cache.

        Parameters
        ----------
        classifications : list[Classification]
            A list of Classification objects.
        show_progress : bool, default=False
            Toggle for tqdm progress bar.
        """

        disable_tqdm = not show_progress
        for classification in tqdm(classifications, disable=disable_tqdm):

            if len(classification.predictions) == 0:
                raise ValueError(
                    "Classifications must contain at least one prediction."
                )

            # update datum uid index
            uid_index = self._add_datum(uid=classification.uid)

            # cache labels and annotations
            groundtruth = self._add_label(classification.groundtruth)

            predictions = list()
            for plabel, pscore in zip(
                classification.predictions, classification.scores
            ):
                label_idx = self._add_label(plabel)
                predictions.append(
                    (
                        label_idx,
                        pscore,
                    )
                )

            pairs = list()
            scores = np.array([score for _, score in predictions])
            max_score_idx = np.argmax(scores)

            for idx, (plabel, score) in enumerate(predictions):
                pairs.append(
                    (
                        float(uid_index),
                        float(groundtruth),
                        float(plabel),
                        float(score),
                        float(max_score_idx == idx),
                    )
                )

            if self._detailed_pairs.size == 0:
                self._detailed_pairs = np.array(pairs)
            else:
                self._detailed_pairs = np.concatenate(
                    [
                        self._detailed_pairs,
                        np.array(pairs),
                    ],
                    axis=0,
                )

    def finalize(self):
        """
        Performs data finalization and some preprocessing steps.

        Returns
        -------
        Evaluator
            A ready-to-use evaluator object.
        """
        if self._detailed_pairs.size == 0:
            raise EmptyEvaluatorError()

        self._label_metadata = compute_label_metadata(
            ids=self._detailed_pairs[:, :3].astype(np.int32),
            n_labels=len(self.index_to_label),
        )
        indices = np.lexsort(
            (
                self._detailed_pairs[:, 1],  # ground truth
                self._detailed_pairs[:, 2],  # prediction
                -self._detailed_pairs[:, 3],  # score
            )
        )
        self._detailed_pairs = self._detailed_pairs[indices]
        self._metadata = Metadata.create(
            detailed_pairs=self._detailed_pairs,
            number_of_datums=len(self.index_to_datum_id),
            number_of_labels=len(self.index_to_label),
        )
        return self

ignored_prediction_labels property

Prediction labels that are not present in the ground truth set.

missing_prediction_labels property

Ground truth labels that are not present in the prediction set.

add_data(classifications, show_progress=False)

Adds classifications to the cache.

Parameters:

Name	Type	Description	Default
classifications	list[Classification]	A list of Classification objects.	required
show_progress	bool	Toggle for tqdm progress bar.	False

Source code in valor_lite/classification/manager.py

def add_data(
    self,
    classifications: list[Classification],
    show_progress: bool = False,
):
    """
    Adds classifications to the cache.

    Parameters
    ----------
    classifications : list[Classification]
        A list of Classification objects.
    show_progress : bool, default=False
        Toggle for tqdm progress bar.
    """

    disable_tqdm = not show_progress
    for classification in tqdm(classifications, disable=disable_tqdm):

        if len(classification.predictions) == 0:
            raise ValueError(
                "Classifications must contain at least one prediction."
            )

        # update datum uid index
        uid_index = self._add_datum(uid=classification.uid)

        # cache labels and annotations
        groundtruth = self._add_label(classification.groundtruth)

        predictions = list()
        for plabel, pscore in zip(
            classification.predictions, classification.scores
        ):
            label_idx = self._add_label(plabel)
            predictions.append(
                (
                    label_idx,
                    pscore,
                )
            )

        pairs = list()
        scores = np.array([score for _, score in predictions])
        max_score_idx = np.argmax(scores)

        for idx, (plabel, score) in enumerate(predictions):
            pairs.append(
                (
                    float(uid_index),
                    float(groundtruth),
                    float(plabel),
                    float(score),
                    float(max_score_idx == idx),
                )
            )

        if self._detailed_pairs.size == 0:
            self._detailed_pairs = np.array(pairs)
        else:
            self._detailed_pairs = np.concatenate(
                [
                    self._detailed_pairs,
                    np.array(pairs),
                ],
                axis=0,
            )

compute_confusion_matrix(score_thresholds=[0.0], hardmax=True, filter_=None)

Computes a detailed confusion matrix..

Parameters:

Name	Type	Description	Default
score_thresholds	list[float]	A list of score thresholds to compute metrics over.	[0.0]
hardmax	bool	Toggles whether a hardmax is applied to predictions.	True
filter_	Filter	Applies a filter to the internal cache.	None

Returns:

Type	Description
list[Metric]	A list of confusion matrices.

Source code in valor_lite/classification/manager.py

def compute_confusion_matrix(
    self,
    score_thresholds: list[float] = [0.0],
    hardmax: bool = True,
    filter_: Filter | None = None,
) -> list[Metric]:
    """
    Computes a detailed confusion matrix..

    Parameters
    ----------
    score_thresholds : list[float]
        A list of score thresholds to compute metrics over.
    hardmax : bool
        Toggles whether a hardmax is applied to predictions.
    filter_ : Filter, optional
        Applies a filter to the internal cache.

    Returns
    -------
    list[Metric]
        A list of confusion matrices.
    """
    # apply filters
    if filter_ is not None:
        detailed_pairs, _ = self.filter(filter_=filter_)
    else:
        detailed_pairs = self._detailed_pairs

    if detailed_pairs.size == 0:
        return list()

    result = compute_confusion_matrix(
        detailed_pairs=detailed_pairs,
        score_thresholds=np.array(score_thresholds),
        hardmax=hardmax,
    )
    return unpack_confusion_matrix_into_metric_list(
        detailed_pairs=detailed_pairs,
        result=result,
        score_thresholds=score_thresholds,
        index_to_datum_id=self.index_to_datum_id,
        index_to_label=self.index_to_label,
    )

compute_precision_recall_rocauc(score_thresholds=[0.0], hardmax=True, filter_=None)

Performs an evaluation and returns metrics.

Parameters:

Name	Type	Description	Default
score_thresholds	list[float]	A list of score thresholds to compute metrics over.	[0.0]
hardmax	bool	Toggles whether a hardmax is applied to predictions.	True
filter_	Filter	Applies a filter to the internal cache.	None

Returns:

Type	Description
dict[MetricType, list]	A dictionary mapping MetricType enumerations to lists of computed metrics.

Source code in valor_lite/classification/manager.py

def compute_precision_recall_rocauc(
    self,
    score_thresholds: list[float] = [0.0],
    hardmax: bool = True,
    filter_: Filter | None = None,
) -> dict[MetricType, list]:
    """
    Performs an evaluation and returns metrics.

    Parameters
    ----------
    score_thresholds : list[float]
        A list of score thresholds to compute metrics over.
    hardmax : bool
        Toggles whether a hardmax is applied to predictions.
    filter_ : Filter, optional
        Applies a filter to the internal cache.

    Returns
    -------
    dict[MetricType, list]
        A dictionary mapping MetricType enumerations to lists of computed metrics.
    """
    # apply filters
    if filter_ is not None:
        detailed_pairs, label_metadata = self.filter(filter_=filter_)
        n_datums = filter_.metadata.number_of_datums
    else:
        detailed_pairs = self._detailed_pairs
        label_metadata = self._label_metadata
        n_datums = self.metadata.number_of_datums

    results = compute_precision_recall_rocauc(
        detailed_pairs=detailed_pairs,
        label_metadata=label_metadata,
        score_thresholds=np.array(score_thresholds),
        hardmax=hardmax,
        n_datums=n_datums,
    )
    return unpack_precision_recall_rocauc_into_metric_lists(
        results=results,
        score_thresholds=score_thresholds,
        hardmax=hardmax,
        label_metadata=label_metadata,
        index_to_label=self.index_to_label,
    )

create_filter(datum_ids=None, labels=None)

Creates a filter object.

Parameters:

Name	Type	Description	Default
datum_uids	list[str]	An optional list of string uids representing datums.	required
labels	list[str]	An optional list of labels.	None

Returns:

Type	Description
Filter	The filter object representing the input parameters.

Source code in valor_lite/classification/manager.py

def create_filter(
    self,
    datum_ids: list[str] | None = None,
    labels: list[str] | None = None,
) -> Filter:
    """
    Creates a filter object.

    Parameters
    ----------
    datum_uids : list[str], optional
        An optional list of string uids representing datums.
    labels : list[str], optional
        An optional list of labels.

    Returns
    -------
    Filter
        The filter object representing the input parameters.
    """
    # create datum mask
    n_pairs = self._detailed_pairs.shape[0]
    datum_mask = np.ones(n_pairs, dtype=np.bool_)
    if datum_ids is not None:
        if not datum_ids:
            return Filter(
                datum_mask=np.zeros_like(datum_mask),
                valid_label_indices=None,
                metadata=Metadata(),
            )
        valid_datum_indices = np.array(
            [self.datum_id_to_index[uid] for uid in datum_ids],
            dtype=np.int32,
        )
        datum_mask = np.isin(
            self._detailed_pairs[:, 0], valid_datum_indices
        )

    # collect valid label indices
    valid_label_indices = None
    if labels is not None:
        if not labels:
            return Filter(
                datum_mask=datum_mask,
                valid_label_indices=np.array([], dtype=np.int32),
                metadata=Metadata(),
            )
        valid_label_indices = np.array(
            [self.label_to_index[label] for label in labels] + [-1]
        )

    filtered_detailed_pairs, _ = filter_cache(
        detailed_pairs=self._detailed_pairs,
        datum_mask=datum_mask,
        valid_label_indices=valid_label_indices,
        n_labels=self.metadata.number_of_labels,
    )

    number_of_datums = (
        len(datum_ids)
        if datum_ids is not None
        else self.metadata.number_of_datums
    )

    return Filter(
        datum_mask=datum_mask,
        valid_label_indices=valid_label_indices,
        metadata=Metadata.create(
            detailed_pairs=filtered_detailed_pairs,
            number_of_datums=number_of_datums,
            number_of_labels=self.metadata.number_of_labels,
        ),
    )

evaluate(score_thresholds=[0.0], hardmax=True, filter_=None)

Computes a detailed confusion matrix..

Parameters:

Name	Type	Description	Default
score_thresholds	list[float]	A list of score thresholds to compute metrics over.	[0.0]
hardmax	bool	Toggles whether a hardmax is applied to predictions.	True
filter_	Filter	Applies a filter to the internal cache.	None

Returns:

Type	Description
dict[MetricType, list[Metric]]	Lists of metrics organized by metric type.

Source code in valor_lite/classification/manager.py

def evaluate(
    self,
    score_thresholds: list[float] = [0.0],
    hardmax: bool = True,
    filter_: Filter | None = None,
) -> dict[MetricType, list[Metric]]:
    """
    Computes a detailed confusion matrix..

    Parameters
    ----------
    score_thresholds : list[float]
        A list of score thresholds to compute metrics over.
    hardmax : bool
        Toggles whether a hardmax is applied to predictions.
    filter_ : Filter, optional
        Applies a filter to the internal cache.

    Returns
    -------
    dict[MetricType, list[Metric]]
        Lists of metrics organized by metric type.
    """
    metrics = self.compute_precision_recall_rocauc(
        score_thresholds=score_thresholds,
        hardmax=hardmax,
        filter_=filter_,
    )
    metrics[MetricType.ConfusionMatrix] = self.compute_confusion_matrix(
        score_thresholds=score_thresholds,
        hardmax=hardmax,
        filter_=filter_,
    )
    return metrics

filter(filter_)

Performs filtering over the internal cache.

Parameters:

Name	Type	Description	Default
filter_	Filter	The filter object representation.	required

Returns:

Type	Description
NDArray[float64]	The filtered detailed pairs.
NDArray[int32]	The filtered label metadata.

Source code in valor_lite/classification/manager.py

def filter(
    self, filter_: Filter
) -> tuple[NDArray[np.float64], NDArray[np.int32]]:
    """
    Performs filtering over the internal cache.

    Parameters
    ----------
    filter_ : Filter
        The filter object representation.

    Returns
    -------
    NDArray[float64]
        The filtered detailed pairs.
    NDArray[int32]
        The filtered label metadata.
    """
    return filter_cache(
        detailed_pairs=self._detailed_pairs,
        datum_mask=filter_.datum_mask,
        valid_label_indices=filter_.valid_label_indices,
        n_labels=self.metadata.number_of_labels,
    )

finalize()

Performs data finalization and some preprocessing steps.

Returns:

Type	Description
Evaluator	A ready-to-use evaluator object.

Source code in valor_lite/classification/manager.py

def finalize(self):
    """
    Performs data finalization and some preprocessing steps.

    Returns
    -------
    Evaluator
        A ready-to-use evaluator object.
    """
    if self._detailed_pairs.size == 0:
        raise EmptyEvaluatorError()

    self._label_metadata = compute_label_metadata(
        ids=self._detailed_pairs[:, :3].astype(np.int32),
        n_labels=len(self.index_to_label),
    )
    indices = np.lexsort(
        (
            self._detailed_pairs[:, 1],  # ground truth
            self._detailed_pairs[:, 2],  # prediction
            -self._detailed_pairs[:, 3],  # score
        )
    )
    self._detailed_pairs = self._detailed_pairs[indices]
    self._metadata = Metadata.create(
        detailed_pairs=self._detailed_pairs,
        number_of_datums=len(self.index_to_datum_id),
        number_of_labels=len(self.index_to_label),
    )
    return self