Added the benchmarking script (Not working properly)

Added metrics for age bracket (`n_off_accuracy`), as well as
binary/multiclass false-{positive,negative} rates (`binary_fpr_fnr`
and `multiclass_fpr_fnr`). Also, added mutual agreement between models
`agreement_fraction` and `agreement_elements`.

.gitignore

@@ -2,4 +2,7 @@
 .vscode
 *.egg-info
 landmark_models
-*__pycache__/
+*__pycache__/
+data/*
+models/*
+gpt*.py

Makefile

@@ -0,0 +1,29 @@
+.PHONY: all clean nuke
+
+all:
+	@echo "Done."
+
+clean:
+	@rm -Rf models/*.pt* models/*.tflite src/facebias/__pycache__ src/facebias/*/__pycache__ src/facebias/estimators/mivolo/__pycache__
+
+nuke: clean
+	@rm -Rf facebias.egg-info src/facebias.egg-info facebias.tar
+
+facebias.tar: nuke
+	tar -cvf facebias.tar src tests Makefile pyproject.toml README.md requirements.txt
+
+models/volo-v1_model_imdb_age_gender_4.22.pth.tar:
+	@mkdir -p models
+	@gdown -O $@ --no-cookies 1NlsNEVijX2tjMe8LBb1rI56WB_ADVHeP
+
+models/fairface_alldata_4race_20191111.pt:
+	@mkdir -p models
+	@gdown -O $@ --no-cookies 1fUJSLseDpgilArB_YKep9PnsR7QrPW5I
+
+models/blaze_face_short_range.tflite:
+	@mkdir -p models
+	@curl --output $@ https://storage.googleapis.com/mediapipe-models/face_detector/blaze_face_short_range/float16/latest/blaze_face_short_range.tflite
+
+models/blaze_face_full_range.tflite:
+	@mkdir -p models
+	@curl --output $@ https://storage.googleapis.com/mediapipe-models/face_detector/blaze_face_full_range/float16/latest/blaze_face_full_range.tflite

README.md

@@ -0,0 +1,6 @@
+# ISR Face Dataset/Model Bias Check API
+
+We use [FairFace](https://github.com/dchen236/FairFace) and [MiVOLO](https://github.com/wildchlamydia/mivolo) version 1, face-only checkpoint.
+
+## Dataset
+Download the tar file `VISTEAM-NAS/Public_Data/facing2-skin-tone-train-images.tar.bz2` to the `data` directory, and extract it. This dataset has balanced sex and skin-tone, and unbalanced age.

benchmark.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import logging
+from pathlib import Path
+
+import mediapipe as mp
+import numpy as np
+import pandas as pd
+import torch.nn as nn
+
+from facebias import load_dataset
+from facebias.detectors import get_face_boxes
+from facebias.detectors.mediapipe import MediapipeDetector
+from facebias.estimators.fairface import FairFace
+from facebias.estimators.mivolov1 import MiVOLOv1
+from facebias.metrics import calc_metrics_per_subgroup, calc_metrics
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(f"facebias:{__file__}")
+
+
+if __name__ == "__main__":
+    import os
+    logger.info(os.getcwd())
+
+    DATASET_PATH = Path("data/facing2-train/")
+    METADATA_PATH = DATASET_PATH / "meta-w-age.csv"
+    DETECTOR_PATH = Path("models/blaze_face_full_range.tflite")
+    # TEST_IMS: list[str] = ["10", "12", "14", "9"]
+
+    detector = MediapipeDetector(str(DETECTOR_PATH))
+    imdict, meta = load_dataset(
+        DATASET_PATH, meta_path=METADATA_PATH, imname_proc_fn=lambda x: x.split("_")[0]
+    )
+    for im, feats in meta.items():
+        age = int(feats["age"])
+        d = age // 10 * 10
+        feats["age_group"] = "{}-{}".format(d, d + 9)
+        feats["age"] = float(feats["age"])
+
+    face_bboxes = get_face_boxes(imdict, detector)
+
+    # for t in TEST_IMS:
+    #     logger.info("-- {} - {}".format(t, meta[str(t)]))
+
+    print(FairFace.capabilities())
+    model_ff = FairFace(Path("models/fairface_alldata_4race_20191111.pt"), device="cpu")
+    preds_ff = model_ff.predict(imdict)
+    metrics_ff = calc_model_performance(meta, preds_ff)
+
+    # logger.info("FairFace -- Test Images")
+    # for t in TEST_IMS:
+    #     logger.info("--{} - {}".format(t, preds_ff[str(t)]))
+
+    metrics_ff_groups = calc_metrics_per_subgroup(meta, preds_ff)
+    for k, v in metrics_ff_groups.items():
+        for kv, vv in v.items():
+            print(k, kv, vv)
+
+    print(MiVOLOv1.capabilities())
+    model_mv = MiVOLOv1(
+        Path("models/volo-v1_model_imdb_age_gender_4.22.pth.tar"), device="cpu"
+    )
+    preds_mv = model_mv.predict(imdict)
+
+    # logger.info("MiVOLOv1(Face Only) -- Test Images")
+    # for t in TEST_IMS:
+    #     logger.info("{} - {}".format(t, preds_mv[str(t)]))
+
+    metrics_mv = calc_model_performance(meta, preds_mv)
+    metrics_mv_groups = calc_metrics_per_subgroup(meta, preds_mv)
+    for k, v in metrics_mv_groups.items():
+        for kv, vv in v.items():
+            print(k, kv, vv)

requirements-dev.txt

@@ -1,6 +0,0 @@
-isort>=8.0.1
-jedi-language-server>=0.46.0
-black>=25.12.0
-flake8>=7.3.0
-mypy>=1.19.0
-pytest>=9.0.0

requirements.txt

@@ -7,3 +7,4 @@ timm==1.0.26
 torch>=2.2.2
 torchvision>=0.17.2
 gdown==6.0.0
+pandas>=3.0.2

src/facebias/__init__.py

@@ -5,13 +5,12 @@ import logging
 from collections import OrderedDict
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Callable, Optional
+from typing import Any, Callable
 
 import cv2
 import numpy as np
 
-logging.basicConfig(level=logging.DEBUG)
-logger = logging.getLogger(__name__)
+logger = logging.getLogger(f"facebias:{__name__}")
 
 
 @dataclass
@@ -22,7 +21,12 @@ class FaceBox:
     y2: int
 
 
-def load_metadata(p: Path, key_id="image", key_proc_fn=None) -> dict[str, dict[str, str]]:
+# TODO(gschardong): Convert all CSV reading functions to pandas
+
+
+def load_metadata(
+    p: Path, key_id="image", key_proc_fn=None
+) -> dict[str, dict[str, str]]:
     lines = []
     with open(p, newline="") as csvfile:
         dialect = csv.Sniffer().sniff(csvfile.read(1024))
@@ -40,10 +44,8 @@ def load_metadata(p: Path, key_id="image", key_proc_fn=None) -> dict[str, dict[s
 
 
 def load_dataset(
-        root: Path,
-        meta_path: Optional[Path] = None,
-        imname_proc_fn: Optional[Callable]=None
-) -> tuple[dict[Path, np.ndarray], Optional[dict[str, dict[str, Any]]]]:
+    root: Path, meta_path: Path | None, imname_proc_fn: Callable | None
+) -> tuple[dict[str, np.ndarray], dict[str, dict[str, Any]] | None]:
     """
     if `meta_path` is `None`, we won't attempt to read it.
 
@@ -53,20 +55,22 @@ def load_dataset(
         Root path to the images. We assume that the images are positioned in a
         flat directory under `root`.
 
-    meta_path: Optional[Path]
+    meta_path: Path | None
         Path to the metadata file on the dataset. Default is `None`.
 
-    imname_proc_fn: Optional[Callable]
+    imname_proc_fn: Callable | None
         Function to apply to the image filenames when building the output
         dictionary. Use it to uniformize it with the metadata file. Default
         is `None`.
 
     Returns
     -------
-    data: dict[Path, np.ndarray]
-    metadata, Optional[dict[str, dict[str, Any]]]
+    data: dict[str, np.ndarray]
+    metadata, dict[str, dict[str, Any]] | None
+        If metadata was not found, or not set to load (by leaving `meta_path`
+        as `None`), then we simply return `None`.
     """
-    metadata = dict()
+    metadata = None
     paths = set([p for p in root.iterdir() if not p.is_dir()])
     if meta_path is not None and meta_path in paths:
         # Just read the metadata and avoid needing to test for it in the main loop.
@@ -74,14 +78,15 @@ def load_dataset(
         paths.remove(meta_path)
 
     ims = OrderedDict()
-    # TODO(gschardong): Paralellize this?
     for p in paths:
         try:
             im = cv2.cvtColor(cv2.imread(p), cv2.COLOR_BGR2RGB)
         except cv2.error:
             logger.info(f'File "{p}" is not an image. Skipping.')
         else:
-            proc_imname = imname_proc_fn(p.name) if imname_proc_fn is not None else p.name
+            proc_imname = (
+                imname_proc_fn(p.name) if imname_proc_fn is not None else str(p.name)
+            )
             ims[proc_imname] = im
 
     if not metadata:
@@ -90,3 +95,157 @@ def load_dataset(
             logger.error(f'Metadata file not found at "{meta_path}".')
 
     return ims, metadata
+
+
+# def calc_model_performance(
+#     gt: pd.DataFrame,
+#     preds: pd.DataFrame,
+#     keys: list[str] | None = None,
+#     possible_caps: dict[Capability, Any] | None = None,
+# ) -> pd.DataFrame:
+#     """
+#     We assume that both `gt` and `preds` have the same structure. They should
+#     be indexed by individual ID, such as the image name, and each value is a
+#     dictionary with model prediction capabilities as keys (e.g., "age_group",
+#     "sex", "skin-color", etc.), and the values are the predictions, or ground-truth
+#     values for each ID/capability.
+
+#     if `keys` is empty, then we infer from common keys present in `preds` and `gt`.
+
+#     Parameters
+#     ----------
+#     gt: pd.DataFrame
+#     preds: pd.DataFrame
+#     keys: list[str] | None
+
+#     Returns
+#     -------
+#     metrics: pd.DataFrame
+#     """
+#     common_caps = keys
+#     if keys is None:
+#         common_caps = set(gt.columns) & set(preds.columns)
+#         if not common_caps:
+#             logger.error(
+#                 f'No common capabilities found. Predictions has "{preds.columns}",'
+#                 f' ground-truth has "{gt.columns}".'
+#             )
+#             return None
+
+#     # Finding common images between predictions and ground-truth.
+#     common_inds = set(preds.index) & set(gt.index)
+#     if not common_inds:
+#         logger.error("No common images found between predictions and ground-truth.")
+#         return None
+
+#     metric_vals = dict()
+#     for cap in common_caps:
+#         if isinstance(preds[cap].iloc[0], (float, int)):
+#             metric_vals[cap] = {
+#                 "mean_absolute_error": mean_absolute_error(gt[cap], preds[cap]),
+#                 "max_error": max_error(gt[cap], preds[cap]),
+#             }
+#         else:
+#             labels = possible_caps[cap]
+#             if possible_caps is None:
+#                 labels = sorted(list(set(preds[cap].unique()) | set(gt[cap].unique())))
+#             metric_vals[cap] = {
+#                 "accuracy": accuracy_score(gt[cap], preds[cap]),
+#                 "balanced_accuracy": balanced_accuracy_score(gt[cap], preds[cap]),
+#                 "cohen-kappa": cohen_kappa_score(gt[cap], preds[cap], labels=labels),
+#             }
+
+#     return pd.DataFrame.from_dict(metric_vals)
+
+
+# def calc_metrics_per_subgroup(
+#     gt: pd.DataFrame,
+#     preds: pd.DataFrame,
+#     model_cls: BaseEstimator,
+#     metrics: list[str] = [
+#         "accuracy",
+#     ],
+# ) -> pd.DataFrame:
+#     """Calculate performance metrics per sub-group for each capability.
+
+#     Parameters
+#     ----------
+#     gt: pd.DataFrame
+
+#     preds: pd.DataFrame
+
+#     model_cls: BaseEstimator-derived class
+
+#     Returns
+#     -------
+#     metrics: dict[Capability, dict[Any, dict]]
+#     """
+#     common_caps = set(gt.columns) & set(preds.columns)
+
+#     # metrics = {}
+#     index = sorted(
+#         [
+#             model_cls.possible_capability_values(c)
+#             for c in common_caps
+#             if c != Capability.AGE
+#         ],
+#         key=len,
+#     )
+#     index = product(*index)
+
+#     metrics = ["accuracy", ""]
+#     df = pd.DataFrame(index=index, columns=metrics)
+
+#     for cap in common_caps:
+#         # TODO(gschardong): Better to store the "type" of each capability
+#         # somewhere and test all numeric types here.
+#         if cap == Capability.AGE:
+#             continue
+
+#         other_caps = common_caps - set([cap])
+
+#         # TODO(gschardong): Do we only need the values that occur in the data,
+#         # or all possible values? If the first is true, then we need to fetch
+#         # from the model class itself, else, we keep it as is.
+#         unique_values_cap = set(gt[cap].unique()) | set(preds[cap].unique())
+
+#         metrics[cap] = {}
+#         for val in unique_values_cap:
+#             ids = gt.index[gt[cap] == val]
+#             metrics[cap][val] = {"number_of_elements": len(ids)}
+#             for ocap in other_caps:
+#                 metrics[cap][val][ocap] = {}
+#                 fpred_data = preds[ocap][ids]
+#                 fgt_data = gt[ocap][ids]
+
+#                 if isinstance(fpred_data[0], (float, int)):
+#                     # If data is numeric, we calculate regression-based metrics
+#                     metrics[cap][val][ocap] = {
+#                         "mean_absolute_error": mean_absolute_error(
+#                             fgt_data, fpred_data
+#                         ),
+#                         "max_error": max_error(fgt_data, fpred_data),
+#                     }
+#                 else:
+#                     unique_values_ocap = sorted(
+#                         list(set(gt[ocap].unique()) | set(preds[ocap].unique()))
+#                     )
+#                     unique_values_ocap = np.array(unique_values_ocap)
+
+#                     metrics_small = {}
+#                     # if len(fgt_data.unique()) == 2:
+#                     #     cm = confusion_matrix(fgt_data, fpred_data, labels=unique_values_ocap)
+#                     # else:
+#                     cm = multilabel_confusion_matrix(
+#                         fgt_data, fpred_data, labels=unique_values_ocap
+#                     )
+#                     for m, oval in zip(cm, unique_values_ocap):
+#                         # metrics[cap][val][ocap][oval] = m
+#                         metrics_small[oval] = m
+#                     return m
+
+#                     metrics[cap][val][ocap] = {
+#                         "accuracy": accuracy_score(fgt_data, fpred_data),
+#                     }
+
+#     return metrics

src/facebias/detectors/__init__.py

@@ -1,8 +1,13 @@
 # -*- coding: utf-8 -*-
 
+import logging
 from abc import ABC, abstractmethod
+from collections import OrderedDict
 
 import numpy as np
+from facebias import FaceBox
+
+logger = logging.getLogger("facebias")
 
 
 class NoFaceDetectedError(Exception):
@@ -20,7 +25,38 @@ class BaseFaceDetector(ABC):
     and perform face detection, whenever a call to `predict` is performed."""
 
     @abstractmethod
-    def detect(self, im) -> np.ndarray:
+    def detect(self, im: np.ndarray) -> np.ndarray:
         """Detects faces in image `im`. Throws `NoFaceDetectedError`
         if `im` has no valid face(s)."""
         pass
+
+
+def get_face_boxes(
+        ims: dict[str, np.ndarray],
+        detector: BaseFaceDetector,
+        num_faces: int=1
+) -> OrderedDict[str, list[FaceBox]]:
+    """
+    Parameters
+    ----------
+    ims: dict[str, np.ndarray],
+    detector: BaseDetector,
+    num_faces: int | 1
+
+    Returns
+    -------
+    all_boxes: OrderedDict[str, list[FaceBox]]
+
+    """
+    all_boxes: OrderedDict[str, list[FaceBox]] = OrderedDict()
+    for p, im in ims.items():
+        boxes = detector.detect(im)
+        if len(boxes) > num_faces:
+            name = "face" if num_faces == 1 else "faces"
+            logger.warning(
+                f"Should only have {num_faces} {name} per image."
+                " Ignoring all others."
+            )
+
+        all_boxes[p] = boxes[:num_faces]
+    return all_boxes

src/facebias/detectors/mediapipe.py

@@ -15,7 +15,7 @@ class MediapipeDetector(BaseFaceDetector):
     Parameters
     ----------
     model_path: str, PathLike
-    mode: Optional[mediapipe.tasks.python.vision.RunningMode]
+    mode: mediapipe.tasks.python.vision.RunningMode
     """
 
     def __init__(self, model_path: str, mode=vision.RunningMode.IMAGE) -> None:

src/facebias/estimators/__init__.py

@@ -12,13 +12,20 @@ class Capability(StrEnum):
     AGE = "age"
     AGEGROUP = "age_group"
     SEX = "sex"
-    SKINCOLOR = "skin_color"
+    SKINTONE = "skin_tone"
     ETHNICITY = "ethinicity"
 
 
+class InvalidCapabilityError(Exception):
+    def __init__(self, msg="Invalid capability for model."):
+        super(InvalidCapability, self).__init__(msg=msg)
+
+
 class BaseEstimator(ABC):
     @abstractmethod
-    def predict(self, images: dict[Path, np.ndarray]) -> dict[Path, dict[Capability, Any]]:
+    def predict(
+        self, images: dict[Path, np.ndarray]
+    ) -> dict[Path, dict[Capability, Any]]:
         """Runs the estimator on a batch of images.
 
         The input `images` is a dictionary indexed by the image name, or ID,
@@ -35,3 +42,8 @@ class BaseEstimator(ABC):
         face image, such as `age_group`, `sex`, `skin_color`, etc.
         """
         pass
+
+    @abstractclassmethod
+    def possible_capability_values(cap: Capability) -> list:
+        """Returns all possible values for a given model capability."""
+        pass

src/facebias/estimators/fairface.py

@@ -2,13 +2,13 @@
 
 from collections import OrderedDict
 from pathlib import Path
-from typing import Any, Optional
+from typing import Any
 
 import numpy as np
 import torch
 import torch.nn as nn
 import torchvision
-from facebias.estimators import BaseEstimator, Capability
+from facebias.estimators import BaseEstimator, Capability, InvalidCapabilityError
 from timm.data import IMAGENET_DEFAULT_MEAN, IMAGENET_DEFAULT_STD
 from torchvision import transforms
 
@@ -24,10 +24,10 @@ class FairFace(BaseEstimator):
 
     Parameters
     ----------
-    checkpoint_path: Optional[Path]
+    checkpoint_path: Path | None
         Path to the model weights stored as a PyTorch file.
 
-    device: Optional[torch.device]
+    device: torch.device
         The device to load the model into. By default is `"cpu"`
 
     Reference
@@ -36,8 +36,8 @@ class FairFace(BaseEstimator):
     """
     def __init__(
             self,
-            checkpoint_path: Optional[Path]=None,
-            device: Optional[torch.device]=torch.device("cpu")
+            checkpoint_path: Path | None,
+            device: torch.device=torch.device("cpu")
     ):
         self.T = transforms.Compose([
             transforms.ToPILImage(),
@@ -70,9 +70,9 @@ class FairFace(BaseEstimator):
             y = y.cpu().detach().squeeze().numpy()
 
             # Ethnicity prediction
-            y_ethno = y[:4]
-            ethno_score = np.exp(y_ethno) / np.sum(np.exp(y_ethno))
-            ethno_pred = np.argmax(ethno_score)
+            # y_ethno = y[:4]
+            # ethno_score = np.exp(y_ethno) / np.sum(np.exp(y_ethno))
+            # ethno_pred = np.argmax(ethno_score)
 
             # Age prediction
             y_age = y[9:18]
@@ -96,9 +96,19 @@ class FairFace(BaseEstimator):
     def capabilities() -> list[Capability]:
         return [Capability.AGEGROUP, Capability.SEX] #, Capability.ETHNICITY]
 
+    def possible_capability_values(cap: Capability) -> list[str]:
+        if cap == Capability.AGEGROUP:
+            return ["00-02", "03-09", "10-19", "20-29", "30-39", "40-49", "50-59", "60-69", "70+"]
+        elif cap == Capability.SEX:
+            return ['m', 'f']
+        # elif cap == Capability.ETHNICITY:
+        #     return ["White", "Black", "Asian", "Indian"]
+        else:
+            raise InvalidCapabilityError()
+
 
 def _to_age_label(age):
-    return "{}-{}".format(age * 10, age * 10 + 9)
+    return FairFace.possible_capability_values(Capability.AGEGROUP)[age]
 
 
 def _to_ethno_label(val):

src/facebias/estimators/mivolo/README.md

@@ -0,0 +1,3 @@
+# Subset of MiVOLO code
+
+This is a subset of the [MiVOLO](https://github.com/wildchlamydia/mivolo) code necessary to instantiate the face-only attribute model.

src/facebias/estimators/mivolov1.py

@@ -2,14 +2,14 @@
 
 from collections import OrderedDict
 from pathlib import Path
-from typing import Any, Optional
+from typing import Any
 
 import numpy as np
 import torch
 import torch.nn as nn
 import torchvision
 from facebias import load_dataset
-from facebias.estimators import BaseEstimator, Capability
+from facebias.estimators import BaseEstimator, Capability, InvalidCapabilityError
 from facebias.estimators.mivolo.mi_volo import MiVOLO
 from timm.data import IMAGENET_DEFAULT_MEAN, IMAGENET_DEFAULT_STD
 from torchvision import transforms
@@ -24,7 +24,7 @@ class MiVOLOv1(BaseEstimator):
         Path to the saved checkpoint. Note that we expect the checkpoint to be
         in PyTorch format.
 
-    device: Optional[torch.device]
+    device: torch.device
         Device to load the checkpoints into. By default is "cpu".
 
     Reference
@@ -34,7 +34,7 @@ class MiVOLOv1(BaseEstimator):
     def __init__(
             self,
             checkpoint_path: Path,
-            device: Optional[torch.device]=torch.device("cpu")
+            device: torch.device=torch.device("cpu")
     ):
         self.T = transforms.Compose([
             transforms.ToPILImage(),
@@ -80,10 +80,28 @@ class MiVOLOv1(BaseEstimator):
     def capabilities() -> list[Capability]:
         return [Capability.AGE, Capability.AGEGROUP, Capability.SEX]
 
+    def possible_capability_values(cap: Capability) -> list[str]:
+        if cap == Capability.AGE:
+            return []
+        elif cap == Capability.AGEGROUP:
+            return ["0-2", "3-9", "10-19", "20-29", "30-39", "40-49", "50-59", "60-69", "70+"]
+        elif cap == Capability.SEX:
+            return ['m', 'f']
+        else:
+            raise InvalidCapabilityError()
+
 
 def _to_age_label(age):
-    base_age = int(age // 10 * 10)
-    return "{}-{}".format(base_age, base_age + 9)
+    iage = int(age)
+    if iage < 3:
+        return "0-2"
+    elif iage < 10:
+        return "3-9"
+    elif iage > 69:
+        return "70+"
+
+    d = int(iage // 10 * 10)
+    return "{}-{}".format(d, d + 9)
 
 
 if __name__ == '__main__':

src/facebias/evaluation.py

@@ -0,0 +1,214 @@
+# -*- coding: utf-8 -*-
+
+"""Dataset/model evaluation functions."""
+
+import logging
+from itertools import permutations, combinations
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+from scipy.stats import entropy
+
+from facebias.estimators import Capability
+from facebias.metrics import gini
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("facebias:evaluation.py")
+
+
+# TODO(gschardong): Move to the same file as `load_dataset`
+def _to_age_bracket(row):
+    iage = int(row["age"])
+    if iage < 3:
+        return "00-02"
+    elif iage < 10:
+        return "03-09"
+    elif iage > 69:
+        return "70+"
+
+    d = iage // 10 * 10
+    return "{}-{}".format(d, d + 9)
+
+
+def generate_description(dataset: pd.DataFrame, name: str | None = None) -> tuple[str]:
+    """Generates a textual description of `dataset`, its variables and values.
+
+    For all supported variables in `dataset`, this function lists their names,
+    types, value ranges and descriptive statistics. It also groups each
+    variable, comparing it to the others, thus creating a subgroup description
+    as well.
+
+    Note that we only support the variables defined by the `Capability` enum.
+
+    Parameters
+    ----------
+    dataset: pd.DataFrame
+
+    name: str, optional
+        The dataset name. Only ever used in the first line of the returned
+        description, for pretty printing.
+
+    Returns
+    -------
+    desc: str
+        The textual description as a human-readable string.
+
+    See Also
+    --------
+    `Capability`
+    """
+    outstr = ""
+    if name is not None:
+        outstr = f"The dataset {name}"
+    else:
+        outstr = f"The dataset"
+
+    outstr += f" has a total of {len(dataset)} {dataset.index.name}s, with the following supported features/capabilities:"
+
+    caps = []
+    for c in Capability:
+        if c.value in dataset:
+            caps.append(c)
+            outstr += f"\n- {c.value}"
+
+    outstr += "\n\nEach feature/capability has the following types and values:"
+    for c in caps:
+        if c == Capability.AGE:
+            outstr += f"\n{c.value}: numeric"
+        else:
+            outstr += f"\n{c.value}: categorical"
+            outstr += f"\n - {sorted(dataset[c].unique())}"
+
+    outstr += "\n\nData distribution statistics."
+    for c in caps:
+        outstr += (
+            f'\nThe feature/capability "{c}" has the following distribution of values:'
+        )
+        if c == Capability.AGE:
+            m1 = dataset[c].min()
+            m2 = dataset[c].max()
+            mean = dataset[c].mean()
+            std = dataset[c].std()
+            p25 = dataset[c].quantile(0.25)
+            median = dataset[c].median()
+            p75 = dataset[c].quantile(0.75)
+
+            outstr += f"\n  - min = {m1}"
+            outstr += f"\n  - max = {m2}"
+            outstr += f"\n  - mean = {mean:.2f}"
+            outstr += f"\n  - std = {std:.2f}"
+            outstr += f"\n  - p25 = {p25}"
+            outstr += f"\n  - p50 = {median}"
+            outstr += f"\n  - p75 = {p75}"
+            outstr += "\n Interqualtile ranges:"
+            outstr += f"\n  - p25-min = {p25 - m1}"
+            outstr += f"\n  - p50-p25 = {median - p25}"
+            outstr += f"\n  - p75-p50 = {p75 - median}"
+            outstr += f"\n  - max-p75 = {m2 - p75}"
+        else:
+            series = dataset[c].value_counts().sort_index()
+            for s in series.index:
+                outstr += f"\n - {s}: {series[s]}"
+
+    outstr += "\n\nPer capability/class data distribution statistics."
+    if Capability.AGE in caps and caps[-1] != Capability.AGE:
+        # Rotating AGE to be the last in the list, as it cannot be the grouping
+        #  variable, since it is numeric.
+        idx = caps.index(Capability.AGE)
+        del caps[idx]
+        caps.append(Capability.AGE)
+        # caps = caps[1:] + caps[:1]
+
+    for c1, c2 in combinations(caps, 2):
+        # Here we test for age, but this should really be a test for any
+        # numerical variable.
+        if c1 == Capability.AGE:
+            continue
+
+        gb = dataset.groupby(c1)[[c2]]
+        tmpdf = None
+        tmpstr = '\nGrouping by "{}", the dataset has the following {} for each value of "{}"'
+
+        if c2 != Capability.AGE:
+            outstr += tmpstr.format(c1, "number of elements", c2)
+            tmpdf = gb.value_counts().sort_index().unstack(level=c1)
+            tmpdf = tmpdf.fillna(0).astype(int)
+            # tmpdf = pd.concat([tmpdf, gb.apply(gini)], axis=1)
+        elif c1 != Capability.AGEGROUP:
+            outstr += tmpstr.format(c1, "statistics", c2)
+            tmpdf = pd.concat(
+                [
+                    gb.min(),
+                    gb.quantile(0.25),
+                    gb.median(),
+                    gb.quantile(0.75),
+                    gb.max(),
+                    gb.apply(gini),
+                ],
+                axis=1,
+            )
+            tmpdf.columns = ["min", "p25", "p50", "p75", "max", "gini_impurity"]
+        else:
+            continue
+
+        outstr += f"\n{tmpdf}\n"
+
+    return outstr
+
+
+if __name__ == "__main__":
+    import os
+
+    logger.info(os.getcwd())
+
+    DATASET_PATH = Path("data/facing2-train/")
+    METADATA_PATH = DATASET_PATH / "meta-w-age.csv"
+
+    meta = pd.read_csv(METADATA_PATH, sep=",", index_col="image")
+    meta[Capability.AGEGROUP.value] = meta.apply(_to_age_bracket, axis=1)
+    meta = meta.sort_index()
+    meta[Capability.AGEGROUP + "_cat"], _ = pd.factorize(
+        meta[Capability.AGEGROUP], sort=True
+    )
+    meta[Capability.SEX + "_cat"], _ = pd.factorize(meta[Capability.SEX], sort=True)
+
+    # GINI IMPURITY
+    # Lower values means a concentration of values around a single class, i.e. bias.
+    age_gini = gini(meta[Capability.AGE])
+
+    agegroup_gini = gini(meta[Capability.AGEGROUP + "_cat"])
+
+    # Should be close to 0.5, indicating a 50/50 split of males and females,
+    # representing maximum uncertainty.
+    sex_gini = gini(meta[Capability.SEX + "_cat"])
+
+    # SHANNON'S ENTROPY
+    count_per_agegroup = meta["age_group"].value_counts()
+    prob_per_agegroup = count_per_agegroup / count_per_agegroup.sum()
+    H_agegroup = entropy(prob_per_agegroup)
+
+    count_per_sex = meta["sex"].value_counts()
+    prob_per_sex = count_per_sex / count_per_sex.sum()
+    H_sex = entropy(prob_per_sex)
+
+    # Now, onto the subgroup metrics.
+    # The goal is to be able to answer the following types of questions:
+    # 1) How many women are in each age-bracket?
+    # 2) Given the population in age-bracket 20-49 years, how is their gender distribution?
+    # 3) Do we need to collect more images of new individuals? If so, what population should we focus on?
+    sex_gb = meta.groupby(Capability.SEX)[["age_group_cat"]]
+    agegroup_gb = meta.groupby(Capability.AGEGROUP)[["sex_cat"]]
+
+    gini_per_sex = sex_gb.apply(gini)
+    gini_per_agegroup = agegroup_gb.apply(gini)
+
+    # Prototype textual description of the dataset.
+    s = generate_description(meta, name=DATASET_PATH.name)
+    print(s)
+
+
+# Diagnostics of biases in the dataset. To be incorporated into a
+# "generate_diagnostics" function later on.
+def generate_bias_diagnostics():
+    pass

src/facebias/metrics.py

@@ -0,0 +1,335 @@
+# -*- coding: utf-8 -*-
+
+"""Model evaluation metrics."""
+
+import logging
+from itertools import combinations
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+import pandas as pd
+from sklearn.metrics import confusion_matrix, multilabel_confusion_matrix
+
+from facebias.estimators import Capability
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("facebias:metrics.py")
+
+
+# Needed for the n_off_accuracy family of functions
+_agegroup_int_map = {
+    "00-02": 0,
+    "03-09": 1,
+    "10-19": 2,
+    "20-29": 3,
+    "30-39": 4,
+    "40-49": 5,
+    "50-59": 6,
+    "60-69": 7,
+    "70+": 8,
+}
+
+
+def n_off_accuracy(gt: list[int], pred: list[int], n: int = 1) -> float:
+    """Returns the n-off accuracy for ordinal class labels encoded as consecutive integers.
+
+    A prediction is counted as correct if it is exact or off by at most `n`.
+    """
+    if len(gt) != len(pred):
+        raise ValueError("Both arrays must have the same length.")
+
+    if len(gt) == 0 or len(pred) == 0:
+        raise ValueError("The arrays must be populated.")
+
+    arr_gt = np.asarray(gt)
+    arr_pred = np.asarray(pred)
+
+    return float(np.mean(np.abs(arr_gt - arr_pred) <= n))
+
+
+def one_off_accuracy(gt: list[int], pred: list[int]) -> float:
+    """1-off accuracy for ordinal class labels encoded as consecutive integers."""
+    return n_off_accuracy(gt, pred, n=1)
+
+
+def two_off_accuracy(gt: list[int], pred: list[int]) -> float:
+    """2-off accuracy for ordinal class labels encoded as consecutive integers."""
+    return n_off_accuracy(gt, pred, n=2)
+
+
+def binary_fpr_fnr(cm: np.ndarray) -> dict[str, np.number]:
+    """Given a confusion matrix, calculates the false-positive and negative rates.
+
+    Parameters
+    ----------
+    cm: np.ndarray
+        The 2x2 confusion matrix returned by `scikit-learn`.
+
+    Returns
+    -------
+    metrics: dict[str, np.number]
+        The calculated metrics keyed under "FPR", "FNR", "TP", "TN", "FP", "FN".
+
+    See Also
+    --------
+    sklearn.metrics.multilabel_confusion_matrix
+    sklearn.metrics.confusion_matrix
+    """
+    tn, fp, fn, tp = cm.ravel()
+    return {
+        "FPR": fp / (fp + tn) if (fp + tn) != 0 else np.nan,
+        "FNR": fn / (fn + tp) if (fn + tp) != 0 else np.nan,
+        "TN": int(tn),
+        "FP": int(fp),
+        "FN": int(fn),
+        "TP": int(tp),
+    }
+
+
+def multiclass_fpr_fnr(
+    gt: pd.Series, preds: pd.Series, labels: list[Any] | None = None
+) -> tuple[dict, list[str]]:
+    """Calculates one-vs-rest false-positive and negative rates for each class.
+
+    Also returns the counts of false-positives, false-negatives, true-positives
+    and true-negatives, i.e., the confusion matrix for each class.
+
+    Parameters
+    ----------
+    gt: pd.Series
+        Indexed ground-truth data.
+
+    preds: pd.Series
+        Indexed predictions.
+
+    labels: list[Any], optional
+        List of labels in the data. If left empty, then we infer from the
+        union of unique elements in `gt` and `preds`.
+
+    Returns
+    -------
+    results: dict[str, dict[str, number]]
+        The false-{positive,negative}-rates, and true/false positive/negative
+        values for each label in `labels`.
+
+    labels: list[str]
+        The labels used when calculating the confusion matrix. If labels was
+        passed as an argument, then it is returned unchanged, else, we return
+        the labels inferred from the data.
+    """
+    if labels is None or not labels:
+        logging.info("Labels not provided. Inferring from the data.")
+        labels = sorted(list(set(gt.unique()) | set(preds.unique())))
+
+    mcm = multilabel_confusion_matrix(gt, preds, labels=labels)
+
+    results = {}
+    for label, m in zip(labels, mcm):
+        results[label] = binary_fpr_fnr(m)
+
+    return results, labels
+
+
+def _agreement_sanity_checks(x_pred: pd.Series, y_pred: pd.Series) -> bool:
+    if len(x_pred) != len(y_pred):
+        raise ValueError(
+            f"Predictions have different lengths. len(x_pred) = {len(x_pred)}"
+            f" len(y_pred) = {len(y_pred)}"
+        )
+    if not all(x_pred.index == y_pred.index):
+        raise ValueError("Index mismatch between series")
+
+    return True
+
+
+def agreement_fraction(x_pred: pd.Series, y_pred: pd.Series) -> float:
+    """Calculates the fraction of agreement between predictions by two models.
+
+    Note that the predictions must both have the same indices and lengths.
+
+    Parameters
+    ----------
+    x_pred: pd.Series
+        Predictions of the first model.
+
+    y_pred: pd.Series
+        Second model predictions.
+
+    Returns
+    -------
+    fraction: float
+        The fraction of agreement between the results.
+    """
+    try:
+        _agreement_sanity_checks(x_pred, y_pred)
+    except ValueError as e:
+        logger.error(f"Cannot calculate agreement fraction -- {str(e)}.")
+        return None
+
+    return (x_pred == y_pred).sum() / len(x_pred)
+
+
+def agreement_elements(
+    x_pred: pd.Series, y_pred: pd.Series, return_disagreement: bool = True
+) -> tuple[pd.Series, pd.Series | None]:
+    """Returns the elements of agreement, and optionally, disagreement between models.
+
+    Note that, as in `agreement_fraction`, the predictions must have the same
+    lengths and matching indices.
+
+    Parameters
+    ----------
+    x_pred: pd.Series
+        Predictions of the first model.
+
+    y_pred: pd.Series
+        Second model predictions.
+
+    return_disagreement: bool, optional
+        Returns the disagreement as well if set (default behavior), otherwise,
+        returns `None` for the `disagreement_idx`.
+
+    Returns
+    -------
+    agreement_idx: pd.Series
+        The index of elements where models have the same results.
+
+    disagreement_idx: pd.Series | None
+        The index of elements where models diverge on their results. Only
+        returned when `return_disagreement` is True, else it returns `None.
+    """
+    try:
+        _agreement_sanity_checks(x_pred, y_pred)
+    except ValueError as e:
+        logger.error(f"Cannot get the (dis)agreement elements -- {str(e)}.")
+        return None
+
+    idx = x_pred.index
+    if return_disagreement:
+        return idx[x_pred == y_pred], idx[x_pred != y_pred]
+    else:
+        return idx[x_pred == y_pred], None
+
+
+def gini(x: list[float]) -> float:
+    x = np.array(x, dtype=np.float32)
+    n = len(x)
+    diffs = sum(abs(i - j) for i, j in combinations(x, r=2))
+    return (diffs / (n**2 * x.mean())).item()
+
+
+# TODO(gschardong): Move to the same file as `load_dataset`
+def _to_age_bracket(row):
+    iage = int(row["age"])
+    if iage < 3:
+        return "00-02"
+    elif iage < 10:
+        return "03-09"
+    elif iage > 69:
+        return "70+"
+
+    d = iage // 10 * 10
+    return "{}-{}".format(d, d + 9)
+
+
+if __name__ == "__main__":
+    import os
+
+    logger.info(os.getcwd())
+
+    from facebias import load_dataset
+    from facebias.detectors import get_face_boxes
+    from facebias.detectors.mediapipe import MediapipeDetector
+    from facebias.estimators.fairface import FairFace
+    from facebias.estimators.mivolov1 import MiVOLOv1
+
+    DATASET_PATH = Path("../../data/facing2-train/")
+    METADATA_PATH = DATASET_PATH / "meta-w-age.csv"
+    DETECTOR_PATH = Path("../../models/blaze_face_full_range.tflite")
+
+    detector = MediapipeDetector(str(DETECTOR_PATH))
+    imdict, _ = load_dataset(
+        DATASET_PATH, meta_path=None, imname_proc_fn=lambda x: x.split("_")[0]
+    )
+    meta = pd.read_csv(METADATA_PATH, sep=",", index_col="image")
+    meta[Capability.AGEGROUP.value] = meta.apply(_to_age_bracket, axis=1)
+    meta = meta.sort_index()
+
+    face_bboxes = get_face_boxes(imdict, detector)
+
+    print(FairFace.capabilities())
+    print(MiVOLOv1.capabilities())
+
+    models = {
+        "fairface": FairFace(
+            Path("../../models/fairface_alldata_4race_20191111.pt"), device="cpu"
+        ),
+        "mivolo": MiVOLOv1(
+            Path("../../models/volo-v1_model_imdb_age_gender_4.22.pth.tar"),
+            device="cpu",
+        ),
+    }
+
+    preds_per_model = dict((k, None) for k in models.keys())
+    for model_name, model in models.items():
+        preds = model.predict(imdict)
+        preds = pd.DataFrame.from_dict(preds).T
+        preds.index.rename("image", inplace=True)
+        preds.index = preds.index.astype(meta.index.dtype)
+        preds = preds.sort_index()
+        preds_per_model[model_name] = preds
+
+    for model_name, preds in preds_per_model.items():
+        gt_age_group_ord = meta["age_group"].apply(lambda x: _agegroup_int_map[x])
+        preds_age_group_ord = preds["age_group"].apply(lambda x: _agegroup_int_map[x])
+
+        acc_one_off = one_off_accuracy(gt_age_group_ord, preds_age_group_ord)
+        acc_two_off = two_off_accuracy(gt_age_group_ord, preds_age_group_ord)
+
+        agegroup_subclass, labels = multiclass_fpr_fnr(
+            meta["age_group"], preds["age_group"]
+        )
+
+        print("==== Age group metrics by class ====")
+        # Print as table.
+        for k, v in agegroup_subclass.items():
+            print(f"Class {k}")
+            for m, vv in v.items():
+                print(f"\t{m} -- {vv}")
+
+        agegroup_subclass = pd.DataFrame.from_dict(agegroup_subclass)
+        print(agegroup_subclass)
+
+        model_cls = type(model)
+        ordered_labels = model_cls.possible_capability_values(Capability.SEX)
+        metrics_sex = binary_fpr_fnr(
+            confusion_matrix(meta["sex"], preds["sex"], labels=ordered_labels)
+        )
+
+        print(
+            "==== Sex metrics ===="
+            f"\nPositive class -- {ordered_labels[0]},"
+            f" Negative class -- {ordered_labels[1]}"
+        )
+        for k, v in metrics_sex.items():
+            print(f"\t{k} -- {v}")
+
+    # Agreement tests
+    model_list = list(models.keys())
+    for i in range(len(model_list)):
+        for j in range(i + 1, len(model_list)):
+            first, second = model_list[i], model_list[j]
+            print(f"{first} -- {second}")
+
+            for cap in model_cls.capabilities():
+                if cap == Capability.AGE:
+                    continue
+                frac = agreement_fraction(
+                    preds_per_model[first][cap], preds_per_model[second][cap]
+                )
+                print(f'Agreement fraction for capability: "{cap}" - {frac}')
+                agreement, disagreement = agreement_elements(
+                    meta[cap], preds[cap], return_disagreement=True
+                )
+                print(disagreement)