tested

Author: Matt Sokoloff

labelbox/data/annotation_types/metrics/confusion_matrix.py

@@ -8,7 +8,6 @@
 
 Count = conint(ge=0, le=1e10)
 
-
 ConfusionMatrixMetricValue = Tuple[Count, Count, Count, Count]
 ConfusionMatrixMetricConfidenceValue = Dict[ConfidenceValue,
                                             ConfusionMatrixMetricValue]

labelbox/data/metrics/confusion_matrix/calculation.py

@@ -1,65 +1,61 @@
-
-
-
-from labelbox.data.metrics.iou.calculation import _mask_iou, miou
+from labelbox.data.metrics.iou.calculation import _get_mask_pairs, _get_vector_pairs, miou
 
 from labelbox.data.annotation_types.metrics.confusion_matrix import \
     ConfusionMatrixMetricValue
 
-
 from labelbox.data.annotation_types.metrics.scalar import ScalarMetricValue
 from typing import List, Optional, Tuple, Union
-from shapely.geometry import Polygon
-from itertools import product
 import numpy as np
 from ...annotation_types import (ObjectAnnotation, ClassificationAnnotation,
-                                 Mask, Geometry, Point, Line, Checklist, Text,
-                                 Radio)
-from ..group import get_feature_pairs, get_identifying_key
+                                 Mask, Geometry, Checklist, Radio)
+from ..processing import get_feature_pairs, get_identifying_key, has_no_annotations, has_no_matching_annotations
 
 
 def confusion_matrix(ground_truths: List[Union[ObjectAnnotation,
-                                           ClassificationAnnotation]],
-                 predictions: List[Union[ObjectAnnotation,
-                                         ClassificationAnnotation]],
-                                        iou: float,
-                 include_subclasses: bool) -> ConfusionMatrixMetricValue:
+                                               ClassificationAnnotation]],
+                     predictions: List[Union[ObjectAnnotation,
+                                             ClassificationAnnotation]],
+                     include_subclasses: bool,
+                     iou: float) -> ConfusionMatrixMetricValue:
 
     annotation_pairs = get_feature_pairs(predictions, ground_truths)
     ious = [
-        feature_confusion_matrix(annotation_pair[0], annotation_pair[1], iou, include_subclasses)
+        feature_confusion_matrix(annotation_pair[0], annotation_pair[1],
+                                 include_subclasses, iou)
         for annotation_pair in annotation_pairs.values()
     ]
     ious = [iou for iou in ious if iou is not None]
 
-    return None if not len(ious) else np.sum(ious, axis = 0 ).tolist()
+    return None if not len(ious) else np.sum(ious, axis=0).tolist()
 
 
-
-def feature_confusion_matrix(ground_truths: List[Union[ObjectAnnotation,
-                                           ClassificationAnnotation]],
-                 predictions: List[Union[ObjectAnnotation,
-                                         ClassificationAnnotation]],
-                                        iou: float,
-                 include_subclasses: bool) -> Optional[ConfusionMatrixMetricValue]:
-    if _no_matching_annotations(ground_truths, predictions):
-        return [0,int(len(predictions) > 0), 0, int(len(ground_truths) > 0)]
-    elif _no_annotations(ground_truths, predictions):
+def feature_confusion_matrix(
+        ground_truths: List[Union[ObjectAnnotation, ClassificationAnnotation]],
+        predictions: List[Union[ObjectAnnotation, ClassificationAnnotation]],
+        include_subclasses: bool,
+        iou: float) -> Optional[ConfusionMatrixMetricValue]:
+    if has_no_matching_annotations(ground_truths, predictions):
+        return [0, int(len(predictions) > 0), 0, int(len(ground_truths) > 0)]
+    elif has_no_annotations(ground_truths, predictions):
         # Note that we could return [0,0,0,0] but that will bloat the imports for no reason
         return None
     elif isinstance(predictions[0].value, Mask):
-        return mask_confusion_matrix(ground_truths, predictions, iou, include_subclasses)
+        return mask_confusion_matrix(ground_truths, predictions, iou,
+                                     include_subclasses)
     elif isinstance(predictions[0].value, Geometry):
-        return vector_confusion_matrix(ground_truths, predictions, iou, include_subclasses)
+        return vector_confusion_matrix(ground_truths, predictions, iou,
+                                       include_subclasses)
     elif isinstance(predictions[0], ClassificationAnnotation):
         return classification_confusion_matrix(ground_truths, predictions)
     else:
         raise ValueError(
             f"Unexpected annotation found. Found {type(predictions[0].value)}")
 
 
-def classification_confusion_matrix(ground_truths: List[ClassificationAnnotation],
-                        predictions: List[ClassificationAnnotation]) -> ConfusionMatrixMetricValue:
+def classification_confusion_matrix(
+        ground_truths: List[ClassificationAnnotation],
+        predictions: List[ClassificationAnnotation]
+) -> ConfusionMatrixMetricValue:
     """
     Computes iou score for all features with the same feature schema id.
 
@@ -70,9 +66,11 @@ def classification_confusion_matrix(ground_truths: List[ClassificationAnnotation
         float representing the iou score for the classification
     """
 
-    if _no_matching_annotations(ground_truths, predictions):
-        return [0,int(len(predictions) > 0), 0, int(len(ground_truths) > 0)]
-    elif _no_annotations(ground_truths, predictions) or len(predictions) > 1 or len(ground_truths) > 1:
+    if has_no_matching_annotations(ground_truths, predictions):
+        return [0, int(len(predictions) > 0), 0, int(len(ground_truths) > 0)]
+    elif has_no_annotations(
+            ground_truths,
+            predictions) or len(predictions) > 1 or len(ground_truths) > 1:
         # Note that we could return [0,0,0,0] but that will bloat the imports for no reason
         return None
 
@@ -91,22 +89,24 @@ def classification_confusion_matrix(ground_truths: List[ClassificationAnnotation
         raise ValueError(f"Unsupported subclass. {prediction}.")
 
 
-
 def vector_confusion_matrix(ground_truths: List[ObjectAnnotation],
-                predictions: List[ObjectAnnotation],
-                iou,
-                include_subclasses: bool,
-                buffer=70.) -> Optional[ConfusionMatrixMetricValue]:
-    if _no_matching_annotations(ground_truths, predictions):
-        return [0,int(len(predictions) > 0), 0, int(len(ground_truths) > 0)]
-    elif _no_annotations(ground_truths, predictions):
+                            predictions: List[ObjectAnnotation],
+                            iou: float,
+                            include_subclasses: bool,
+                            buffer=70.) -> Optional[ConfusionMatrixMetricValue]:
+    if has_no_matching_annotations(ground_truths, predictions):
+        return [0, int(len(predictions) > 0), 0, int(len(ground_truths) > 0)]
+    elif has_no_annotations(ground_truths, predictions):
         return None
 
     pairs = _get_vector_pairs(ground_truths, predictions, buffer=buffer)
     return object_pair_confusion_matrix(pairs, iou, include_subclasses)
 
 
-def object_pair_confusion_matrix(pairs : List[Tuple[ObjectAnnotation, ObjectAnnotation, ScalarMetricValue]], iou, include_subclasses) -> ConfusionMatrixMetricValue:
+def object_pair_confusion_matrix(
+        pairs: List[Tuple[ObjectAnnotation, ObjectAnnotation,
+                          ScalarMetricValue]], iou,
+        include_subclasses) -> ConfusionMatrixMetricValue:
     pairs.sort(key=lambda triplet: triplet[2], reverse=True)
     prediction_ids = set()
     ground_truth_ids = set()
@@ -122,9 +122,12 @@ def object_pair_confusion_matrix(pairs : List[Tuple[ObjectAnnotation, ObjectAnno
         if agreement > iou and \
          prediction_id not in matched_predictions and \
          ground_truth_id not in matched_ground_truths:
-            if include_subclasses and (ground_truth.classifications or prediction.classifications):
-                if miou(prediction.classifications, ground_truth.classifications) < 1.:
-                    # Incorrect if the subclasses don't 100% agree
+            if include_subclasses and (ground_truth.classifications or
+                                       prediction.classifications):
+                if miou(prediction.classifications,
+                        ground_truth.classifications,
+                        include_subclasses=False) < 1.:
+                    # Incorrect if the subclasses don't 100% agree then there is no match
                     continue
             matched_predictions.add(prediction_id)
             matched_ground_truths.add(ground_truth_id)
@@ -136,89 +139,56 @@ def object_pair_confusion_matrix(pairs : List[Tuple[ObjectAnnotation, ObjectAnno
     return [tps, fps, tns, fns]
 
 
-
-def _get_vector_pairs(
-        ground_truths: List[ObjectAnnotation],
-        predictions: List[ObjectAnnotation], buffer: float
-) -> List[Tuple[ObjectAnnotation, ObjectAnnotation, ScalarMetricValue]]:
-    """
-    # Get iou score for all pairs of ground truths and predictions
-    """
-    pairs = []
-    for prediction, ground_truth in product(predictions, ground_truths):
-        if isinstance(prediction.value, Geometry) and isinstance(
-                ground_truth.value, Geometry):
-            if isinstance(prediction.value, (Line, Point)):
-                score = _polygon_iou(prediction.value.shapely.buffer(buffer),
-                                     ground_truth.value.shapely.buffer(buffer))
-            else:
-                score = _polygon_iou(prediction.value.shapely,
-                                     ground_truth.value.shapely)
-            pairs.append((prediction, ground_truth, score))
-    return pairs
-
-def _get_mask_pairs(
-        ground_truths: List[ObjectAnnotation],
-        predictions: List[ObjectAnnotation]
-) -> List[Tuple[ObjectAnnotation, ObjectAnnotation, ScalarMetricValue]]:
-    """
-    # Get iou score for all pairs of ground truths and predictions
-    """
-    pairs = []
-    for prediction, ground_truth in product(predictions, ground_truths):
-        if isinstance(prediction.value, Mask) and isinstance(
-                ground_truth.value, Mask):
-            score = _mask_iou(prediction.value.draw(color = 1),
-                                     ground_truth.value.draw(color = 1))
-            pairs.append((prediction, ground_truth, score))
-    return pairs
-
-def _polygon_iou(poly1: Polygon, poly2: Polygon) -> ScalarMetricValue:
-    """Computes iou between two shapely polygons."""
-    if poly1.intersects(poly2):
-        return poly1.intersection(poly2).area / poly1.union(poly2).area
-    return 0.
-
-
-def radio_confusion_matrix(ground_truth: Radio, prediction: Radio) -> ScalarMetricValue:
+def radio_confusion_matrix(ground_truth: Radio,
+                           prediction: Radio) -> ConfusionMatrixMetricValue:
     """
     Calculates confusion between ground truth and predicted radio values
 
     The way we are calculating confusion matrix metrics:
         - TNs aren't defined because we don't know how many other classes exist ... etc
 
-    We treat each example as 1 vs all
+    When P == L, then we get [1,0,0,0]
+    when P != L, we get [0,1,0,1]
 
+    This is because we are aggregating the stats for the entire radio. Not for each class.
+    Since we are not tracking TNs (P == L) only adds to TP.
+    We are not tracking TNs because the number of TNs is equal to the number of classes which we do not know
+    from just looking at the predictions and labels. Also TNs are necessary for precision/recall/f1.
     """
     key = get_identifying_key([prediction.answer], [ground_truth.answer])
     prediction_id = getattr(prediction.answer, key)
     ground_truth_id = getattr(ground_truth.answer, key)
 
+    if prediction_id == ground_truth_id:
+        return [1, 0, 0, 0]
+    else:
+        return [0, 1, 0, 1]
 
 
-
-    return float(getattr(prediction.answer, key) ==
-                 getattr(ground_truth.answer, key))
-
-
-
-
-def checklist_confusion_matrix(ground_truth: Checklist, prediction: Checklist) -> ScalarMetricValue:
+def checklist_confusion_matrix(
+        ground_truth: Checklist,
+        prediction: Checklist) -> ConfusionMatrixMetricValue:
     """
     Calculates agreement between ground truth and predicted checklist items
+
+    Also not tracking TNs
     """
     key = get_identifying_key(prediction.answer, ground_truth.answer)
     schema_ids_pred = {getattr(answer, key) for answer in prediction.answer}
-    schema_ids_label = {
-        getattr(answer, key) for answer in ground_truth.answer
-    }
-    return float(
-        len(schema_ids_label & schema_ids_pred) /
-        len(schema_ids_label | schema_ids_pred))
+    schema_ids_label = {getattr(answer, key) for answer in ground_truth.answer}
+    agree = schema_ids_label & schema_ids_pred
+    all_selected = schema_ids_label | schema_ids_pred
+    disagree = all_selected.difference(agree)
+    fps = len({x for x in disagree if x in schema_ids_pred})
+    fns = len({x for x in disagree if x in schema_ids_label})
+    tps = len(agree)
+    return [tps, fps, 0, fns]
 
 
-def mask_confusion_matrix(ground_truths: List[ObjectAnnotation],
-              predictions: List[ObjectAnnotation], iou, include_subclasses: bool) -> Optional[ScalarMetricValue]:
+def mask_confusion_matrix(
+        ground_truths: List[ObjectAnnotation],
+        predictions: List[ObjectAnnotation], iou,
+        include_subclasses: bool) -> Optional[ScalarMetricValue]:
     """
     Computes iou score for all features with the same feature schema id.
     Calculation includes subclassifications.
@@ -229,9 +199,9 @@ def mask_confusion_matrix(ground_truths: List[ObjectAnnotation],
     Returns:
         float representing the iou score for the masks
     """
-    if _no_matching_annotations(ground_truths, predictions):
-        return [0,int(len(predictions) > 0), 0, int(len(ground_truths) > 0)]
-    elif _no_annotations(ground_truths, predictions):
+    if has_no_matching_annotations(ground_truths, predictions):
+        return [0, int(len(predictions) > 0), 0, int(len(ground_truths) > 0)]
+    elif has_no_annotations(ground_truths, predictions):
         return None
 
     if include_subclasses:
@@ -240,7 +210,8 @@ def mask_confusion_matrix(ground_truths: List[ObjectAnnotation],
         # Otherwise this will flatten the masks.
         # TODO: Make this more apprent in the configuration.
         pairs = _get_mask_pairs(ground_truths, predictions)
-        return object_pair_confusion_matrix(pairs, iou, include_subclasses=include_subclasses)
+        return object_pair_confusion_matrix(
+            pairs, iou, include_subclasses=include_subclasses)
 
     prediction_np = np.max([pred.value.draw(color=1) for pred in predictions],
                            axis=0)
@@ -253,27 +224,7 @@ def mask_confusion_matrix(ground_truths: List[ObjectAnnotation],
             f" Found {prediction_np.shape}/{ground_truth_np.shape}.")
 
     tp_mask = prediction_np == ground_truth_np == 1
-    fp_mask = (prediction_np == 1) & (ground_truth_np==0)
-    fn_mask = (prediction_np == 0) & (ground_truth_np==1)
+    fp_mask = (prediction_np == 1) & (ground_truth_np == 0)
+    fn_mask = (prediction_np == 0) & (ground_truth_np == 1)
     tn_mask = prediction_np == ground_truth_np == 0
     return [np.sum(tp_mask), np.sum(fp_mask), np.sum(fn_mask), np.sum(tn_mask)]
-
-
-
-def _no_matching_annotations(ground_truths: List[ObjectAnnotation],
-                             predictions: List[ObjectAnnotation]):
-    if len(ground_truths) and not len(predictions):
-        # No existing predictions but existing ground truths means no matches.
-        return True
-    elif not len(ground_truths) and len(predictions):
-        # No ground truth annotations but there are predictions means no matches
-        return True
-    return False
-
-
-def _no_annotations(ground_truths: List[ObjectAnnotation],
-                    predictions: List[ObjectAnnotation]):
-    return not len(ground_truths) and not len(predictions)
-
-
-