Added the possible_capability_values base method for estimators...

New `possible_capability_values` base method for estimators. The goal
is to return a list of possible values for each non-numerical
capability, as some performance metrics may benefit from this complete
listing.

Additionally, added a new exception to indicate when a capability is
invalid in a given context, e.g., when the model does not support it.

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

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,7 +5,7 @@ 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
@@ -41,9 +41,9 @@ 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,
+        meta_path: Path | None,
-        imname_proc_fn: Optional[Callable]=None
+        imname_proc_fn: Callable |None
-) -> tuple[dict[Path, np.ndarray], Optional[dict[str, dict[str, Any]]]]:
+) -> 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 +53,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]
+    data: dict[str, np.ndarray]
-    metadata, Optional[dict[str, dict[str, Any]]]
+    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 +76,13 @@ 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:

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

@@ -16,6 +16,11 @@ class Capability(StrEnum):
     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]]:
@@ -35,3 +40,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,
+            checkpoint_path: Path | None,
-            device: Optional[torch.device]=torch.device("cpu")
+            device: torch.device=torch.device("cpu")
     ):
         self.T = transforms.Compose([
             transforms.ToPILImage(),
@@ -96,9 +96,20 @@ class FairFace(BaseEstimator):
     def capabilities() -> list[Capability]:
         return [Capability.AGEGROUP, Capability.SEX] #, Capability.ETHNICITY]
 
+    def possible_capability_values(cap: Capability) -> list:
+        if 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']
+        # 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_classes(Capability.AGEGROUP)[age]
+    # return "{}-{}".format(age * 10, age * 10 + 9)
 
 
 def _to_ethno_label(val):

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)
+    iage = int(age)
-    return "{}-{}".format(base_age, base_age + 9)
+    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/metrics.py

@@ -0,0 +1,289 @@
+# -*- coding: utf-8 -*-
+
+import logging
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+import pandas as pd
+from sklearn.metrics import (accuracy_score, cohen_kappa_score,
+                             confusion_matrix, max_error, mean_absolute_error,
+                             multilabel_confusion_matrix, balanced_accuracy_score)
+
+from facebias.estimators import Capability
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(f"facebias:{__name__}")
+
+
+def calc_model_performance(
+        gt: pd.DataFrame,
+        preds: pd.DataFrame,
+        keys: list[str] = [],
+) -> 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: dict[str, dict[str, Any]]
+    preds: dict[str, dict[str, Any]]
+    keys: list[str] | None
+
+    Returns
+    -------
+    metrics: dict[str, dict[str, float]]
+    """
+    common_caps = keys
+    if not keys:
+        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 = 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 _find_unique_values_per_capability(
+    class_output: dict[str, dict[Capability, Any]], caps: list[Capability] | None = None
+) -> dict[Capability, str]:
+    """Returns the set of values per capability in `class_output`.
+
+    Parameters
+    ----------
+    class_output: dict[str, dict[Capability, Any]]
+        The classification results, or ground-truth data indexed by element.
+
+    caps: list[Capability] | None
+        The list of capabilities to find unique values for. If left as `None`,
+        we will find unique values for all of them.
+
+    Results
+    -------
+    unique_vals: dict[Capability, str]
+        The unique values indexed by capability.
+    """
+    if caps is None:
+        caps = list(next(iter(class_output.values())).keys())
+    elif not isinstance(caps, (list, tuple)):
+        caps = [caps]
+
+    unique_vals = dict()
+    for cap in caps:
+        unique_vals[cap] = set()
+        for res in class_output.values():
+            unique_vals[cap].add(res[cap])
+
+    return unique_vals
+
+
+def _get_capability_data(
+    class_outputs: dict[str, dict[Capability, Any]], cap: Capability
+) -> dict[str, Any]:
+    """Returns data for all individuals regarding a capability.
+
+    Parameters
+    ----------
+    class_outputs: dict[str, dict[Capability, Any]]
+        The estimator outputs indexed by individual.
+
+    cap: Capability
+        The desired capability.
+
+    Returns
+    -------
+    data: dict[str, Any]
+        The capability data indexed by individual.
+    """
+    data_per_id = dict()
+
+    for ind, data in class_outputs.items():
+        if cap not in data:
+            logger.warning(
+                f'Entry for capability "{cap.value}" not found for individual "{ind}". Skipping.'
+            )
+            continue
+        data_per_id[ind] = data[cap]
+
+    return data_per_id
+
+
+def _filter_by_index(data: dict[str, Any], indx: Any):
+    return dict((k, v) for k, v in data.items() if k in indx)
+
+
+def calc_metrics_per_subgroup(
+    gt: dict[str, dict[str, Any]], preds: dict[str, dict[str, Any]]
+) -> dict[Capability, dict[Any, dict]]:
+    """Calculate performance metrics per sub-group for each capability.
+
+    Parameters
+    ----------
+    gt: dict[str, dict[str, Any]]
+
+    preds: dict[str, dict[str, Any]]
+
+    Returns
+    -------
+    metrics: dict[Capability, dict[Any, dict]]
+    """
+    common_caps = set(_find_common_capabilities(gt, preds))
+
+    metrics = {}
+    for cap in common_caps:
+        if cap == Capability.AGE:
+            continue
+
+        other_caps = common_caps - set([cap])
+        unique_values_cap = _find_unique_values_per_capability(gt, cap)[cap]
+
+        metrics[cap] = {}
+        for val in unique_values_cap:
+            ids = [k for k, v in gt.items() if v[cap] == val]
+
+            metrics[cap][val] = {"number_of_elements": len(ids)}
+            for ocap in other_caps:
+                metrics[cap][val][ocap] = {}
+                filtered_pred = _filter_by_index(_get_capability_data(preds, ocap), ids)
+                filtered_gt = _filter_by_index(_get_capability_data(gt, ocap), ids)
+
+                filtered_pred_data = np.array([filtered_pred[i] for i in ids])
+                filtered_gt_data = np.array([filtered_gt[i] for i in ids])
+
+                if isinstance(filtered_pred_data[0], float):
+                    # If data is numeric, we calculate regression-based metrics
+                    metrics[cap][val][ocap] = {
+                        "mean_absolute_error": mean_absolute_error(
+                            filtered_gt_data, filtered_pred_data
+                        ),
+                        "max_error": max_error(filtered_gt_data, filtered_pred_data),
+                    }
+                else:
+                    if len(np.unique(filtered_gt_data)) == 2:
+                        print(cap, val, ocap)
+                        cm = confusion_matrix(filtered_gt_data, filtered_pred_data, labels=get_unique_labels(filtered_gt_data, ocap))
+                    else:
+                        print(cap, val, ocap)
+                        cm = multilabel_confusion_matrix(filtered_gt_data, filtered_pred_data, labels=get_unique_labels(filtered_gt, ocap))
+                    return cm
+
+                    metrics[cap][val][ocap] = {
+                        "accuracy": accuracy_score(
+                            filtered_gt_data, filtered_pred_data
+                        ),
+                    }
+
+    return metrics
+
+
+def get_unique_labels(data, cap: Capability):
+    caps = set()
+    for id, vals in data.items():
+        if cap not in vals:
+            logger.warning(f'Capability "{cap.value}" not registered for individual "{id}". Skipping')
+            continue
+
+        caps.add(vals[cap])
+
+    return caps
+
+
+def _to_age_bracket(row):
+    iage = int(row["age"])
+    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")
+    # TEST_IMS: list[str] = ["10", "12", "14", "9"]
+
+    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)
+
+    # 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)
+    preds_ff = pd.DataFrame.from_dict(preds_ff).T
+    preds_ff.index.rename("image", inplace=True)
+    preds_ff.index = preds_ff.index.astype(meta.index.dtype)
+    preds_ff = preds_ff.sort_index()
+
+    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)
+    preds_mv = pd.DataFrame.from_dict(preds_mv).T
+    preds_mv.index.rename("image", inplace=True)
+    preds_mv.index = preds_mv.index.astype(meta.index.dtype)
+    preds_mv = preds_mv.sort_index()
+
+    # 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)