Added the benchmarking script (Not working properly)

.gitignore

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

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)

src/facebias/__init__.py

@@ -23,7 +23,10 @@ class FaceBox:
 
 # 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]]:
+
+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))
@@ -41,9 +44,7 @@ def load_metadata(p: Path, key_id="image", key_proc_fn=None) -> dict[str, dict[s
 
 
 def load_dataset(
-        root: Path,
-        meta_path: Path | None,
-        imname_proc_fn: Callable |None
+    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.
@@ -83,7 +84,9 @@ def load_dataset(
         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 str(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:
@@ -92,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/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/evaluation.py

@@ -31,12 +31,138 @@ def _to_age_bracket(row):
     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/")
+    DATASET_PATH = Path("data/facing2-train/")
     METADATA_PATH = DATASET_PATH / "meta-w-age.csv"
 
     meta = pd.read_csv(METADATA_PATH, sep=",", index_col="image")
@@ -49,9 +175,8 @@ if __name__ == "__main__":
 
     # GINI IMPURITY
     # Lower values means a concentration of values around a single class, i.e. bias.
-    age_gini = gini(meta["age"])
+    age_gini = gini(meta[Capability.AGE])
 
-    # gt_age_group_ord =  meta["age_group"].apply(lambda x: _agegroup_int_map[x])
     agegroup_gini = gini(meta[Capability.AGEGROUP + "_cat"])
 
     # Should be close to 0.5, indicating a 50/50 split of males and females,
@@ -78,64 +203,12 @@ if __name__ == "__main__":
     gini_per_sex = sex_gb.apply(gini)
     gini_per_agegroup = agegroup_gb.apply(gini)
 
-    # Prototype textual description of the dataset. To be incorporated into a
-    # "generate_report" function.
-    print(
-        f'The dataset "{DATASET_PATH.name}" has a total of {len(meta)} {meta.index.name}s,'
-        " with the following features/capabilities:"
-    )
-    caps = []
-    for c in Capability:
-        if c.value in meta:
-            caps.append(c)
-            print(f"- {c.value}")
+    # Prototype textual description of the dataset.
+    s = generate_description(meta, name=DATASET_PATH.name)
+    print(s)
 
-    print("\nEach feature/capability has the following types and values:")
-    for c in caps:
-        if c == Capability.AGE:
-            print(f"{c.value}: numeric")
 
-        else:
-            print(f"{c.value}: categorical")
-            print(f" - {sorted(meta[c].unique())}")
-
-    print("\nData distribution statistics.")
-    for c in caps:
-        print(f'The feature/capability "{c}" has the following distribution of values:')
-        if c == Capability.AGE:
-            m1 = meta[c].min()
-            m2 = meta[c].max()
-            mean = meta[c].mean()
-            std = meta[c].std()
-            p25 = meta[c].quantile(0.25)
-            median = meta[c].median()
-            p75 = meta[c].quantile(0.75)
-
-            print(
-                f"  - min = {m1}, max = {m2}, mean = {mean:.2f}, std = {std:.2f}"
-                f" p25 = {p25}, p50 = {median}, p75 = {p75}"
-            )
-            print(" Interqualtile ranges:")
-            print(f"  - p25-min = {p25 - m1}")
-            print(f"  - p50-p25 = {median - p25}")
-            print(f"  - p75-p50 = {p75 - median}")
-            print(f"  - max-p75 = {m2 - p75}")
-        else:
-            series = meta[c].value_counts().sort_index()
-            for s in series.index:
-                print(f" - {s}: {series[s]}")
-
-    print("\nPer capability/class data distribution statistics.")
-    for c1, c2 in combinations(caps, 2):
-        if c1 == Capability.AGE:
-            continue
-
-        if c2 != Capability.AGE:
-            gb = meta.groupby(c1)[[c2]]
-            print(
-                f'Grouping by "{c1}", the dataset has the following data distribution for "{c2}"'
-            )
-            print(gb.value_counts().sort_index().unstack(level=c1).fillna(0))
-
-    # Diagnostics of biases in the dataset. To be incorporated into a
-    # "generate_diagnostics" function later on.
+# Diagnostics of biases in the dataset. To be incorporated into a
+# "generate_diagnostics" function later on.
+def generate_bias_diagnostics():
+    pass