nb_projects.robustness_benchmark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# default_exp projects.robustness_benchmark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Robustness Benchmark\n",
    "> Benchmark utility code"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# hide\n",
    "from nbdev.showdoc import *\n",
    "from fastcore.test import *\n",
    "\n",
    "%load_ext autoreload\n",
    "%autoreload 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# export\n",
    "import os\n",
    "import torch\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "\n",
    "from enum import Enum \n",
    "from collections import Counter\n",
    "from functools import partial\n",
    "from scp.analysis.binary import performance\n",
    "from scp.utils.dict import apply_to_vals, flatten_intra"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Constants"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# export\n",
    "Corruption = Enum(\"Corruption\", \n",
    "                  \"gaussian_noise, shot_noise, impulse_noise,\\\n",
    "                   defocus_blur, motion_blur, zoom_blur,\\\n",
    "                   black_corner, char,\\\n",
    "                   brightness_up, brightness_down, contrast, elastic_transform, pixelate, jpeg_compression,\\\n",
    "                   speckle_noise, gaussian_blur, bubble, saturate,\\\n",
    "                   sanity_check\")\n",
    "\n",
    "Perturbation = Enum(\"Perturbation\",\n",
    "                    \"gaussian_noise, shot_noise,\\\n",
    "                     motion_blur, zoom_blur,\\\n",
    "                     char\\\n",
    "                     brightness, translate, rotate, tilt, scale,\\\n",
    "                     speckle_noise, gaussian_blur, bubble, shear,\\\n",
    "                     sanity_check\")\n",
    "\n",
    "max_severity = 6 # severity ranges from 1 to 5 inclusive\n",
    "num_frames = 31 # number of perturbation frames\n",
    "classes = [\"nevus\", \"melanoma\"] # diagnostic classes\n",
    "\n",
    "# AlexNet baseline (unadjusted o.c. and averaged over 5 identical runs)\n",
    "# requires python >= 3.9\n",
    "class BenchmarkBaseline():\n",
    "    '''AlexNet baseline'''\n",
    "    \n",
    "    # SAM\n",
    "    sam = pd.DataFrame({'clean': 0.20590927835051548}, index=[0])\n",
    "    \n",
    "    # SAM-C\n",
    "    samc = pd.DataFrame({'gaussian_noise'   : 0.4031661855670104,\n",
    "             'shot_noise'       : 0.3797690721649484,\n",
    "             'impulse_noise'    : 0.4142465979381443,\n",
    "             'defocus_blur'     : 0.26223587628865985,\n",
    "             'motion_blur'      : 0.2529162886597938,\n",
    "             'zoom_blur'        : 0.35642556701030925,\n",
    "             'black_corner'     : 0.2864305154639175,\n",
    "             'char'             : 0.21900288659793812,\n",
    "             'brightness_up'    : 0.2397558762886598,\n",
    "             'brightness_down'  : 0.32653938144329897,\n",
    "             'contrast'         : 0.4829649484536082,\n",
    "             'elastic_transform': 0.21677113402061854,\n",
    "             'pixelate'         : 0.21391587628865974,\n",
    "             'jpeg_compression' : 0.2977707216494845}, index=[0]) \n",
    "    \n",
    "    # SAM-C-Extra\n",
    "    samce = pd.DataFrame({'speckle_noise': 0.3187744329896908,\n",
    "              'gaussian_blur': 0.2505723711340206,\n",
    "              'bubble'       : 0.401682474226804,\n",
    "              'saturate'     : 0.335340206185567}, index=[0])\n",
    "    \n",
    "    # SAM-P\n",
    "    samp = pd.DataFrame({'gaussian_noise': 0.3045558989048004,\n",
    "             'shot_noise'    : 0.2891536056995392,\n",
    "             'motion_blur'   : 0.010909090749919416,\n",
    "             'zoom_blur'     : 0.018934169411659242,\n",
    "             'char'          : 0.023134795948863033,\n",
    "             'brightness'    : 0.01862068921327591,\n",
    "             'translate'     : 0.023343783244490626,\n",
    "             'rotate'        : 0.04443051218986512,\n",
    "             'tilt'          : 0.03381400182843208,\n",
    "             'scale'         : 0.04390804693102837}, index=[0])\n",
    "    \n",
    "    # SAM-P-Extra\n",
    "    sampe =  pd.DataFrame({'speckle_noise': 0.2331452488899231,\n",
    "               'gaussian_blur': 0.005015673954039813,\n",
    "               'bubble'       : 0.02865203730762005,\n",
    "               'shear'        : 0.038390804082155235}, index=[0])\n",
    "    \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>clean</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0.205909</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "      clean\n",
       "0  0.205909"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "BenchmarkBaseline.sam"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Analysis"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# export\n",
    "# get balanced error rate\n",
    "def bal_error_rate(v):\n",
    "    perf = performance(v[1], v[0][:,1], bal_err_rate=True)\n",
    "    return perf[\"bal_err_rate\"]\n",
    "\n",
    "# get normal error rate\n",
    "def error_rate(v):\n",
    "    perf = performance(v[1], v[0][:,1], err_rate=True)\n",
    "    return perf[\"err_rate\"]\n",
    "\n",
    "# get auroc\n",
    "def auroc(v):\n",
    "    perf = performance(v[1], v[0][:,1], auroc=True)\n",
    "    return perf[\"auroc\"]\n",
    "\n",
    "# get acc\n",
    "def acc(v):\n",
    "    perf = performance(v[1], v[0][:,1])\n",
    "    return perf[\"acc\"]\n",
    "\n",
    "# get spec\n",
    "def spec(v):\n",
    "    perf = performance(v[1], v[0][:,1])\n",
    "    return perf[\"sens\"]\n",
    "\n",
    "# get sens\n",
    "def sens(v):\n",
    "    perf = performance(v[1], v[0][:,1])\n",
    "    return perf[\"spec\"]\n",
    "\n",
    "# get sens\n",
    "def bal_acc(v):\n",
    "    perf = performance(v[1], v[0][:,1], bal_acc=True)\n",
    "    return perf[\"bal_acc\"]\n",
    "\n",
    "def flip_rate(k, v, n):\n",
    "    preds = v[0]\n",
    "    stop = preds.shape[0]\n",
    "    assert stop%n == 0, f\"Number of predictions ({stop}) is not evenly divisible by stepsize ({n})!\"\n",
    "    \n",
    "    noise_seq = False\n",
    "    if \"noise\" in k:\n",
    "        noise_seq = True\n",
    "        \n",
    "    # calculate flip rate for a single image sequence (which contains 'n' number of frames)\n",
    "    n_sum, m = 0, 0\n",
    "    for i in range(0, stop, n):\n",
    "        m += 1\n",
    "        seq_preds = preds[i:i+n].argmax(dim=1)\n",
    "        \n",
    "        if noise_seq:\n",
    "            n_sum += sum(seq_preds[0]!=seq_preds[1:])\n",
    "        else:\n",
    "            n_sum += sum(seq_preds[:-1]!=seq_preds[1:])\n",
    "    \n",
    "    return (n_sum/(m*(n-1))).item()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# export\n",
    "\n",
    "# get flib probability\n",
    "def flip_prob(k, v, n):\n",
    "    preds = v[0]\n",
    "    stop = preds.shape[0]\n",
    "    assert stop%n == 0, f\"Number of predictions ({stop}) is not evenly divisible by stepsize ({n})!\"\n",
    "    \n",
    "    noise_seq = False\n",
    "    if \"noise\" in k:\n",
    "        noise_seq = True\n",
    "        \n",
    "    # calculate flip rate for a single image sequence (which contains 'n' number of frames)\n",
    "    n_sum, m = 0, 0\n",
    "    for i in range(0, stop, n):\n",
    "        m += 1\n",
    "        seq_preds = preds[i:i+n].argmax(dim=1)\n",
    "        \n",
    "        if noise_seq:\n",
    "            n_sum += sum(seq_preds[0]!=seq_preds[1:])\n",
    "        else:\n",
    "            n_sum += sum(seq_preds[:-1]!=seq_preds[1:])\n",
    "    \n",
    "    return (n_sum/(m*(n-1))).item()\n",
    "\n",
    "def relative_perf(perf_c:dict, perf_cl:float):\n",
    "    \n",
    "    def subtract(v, x): return v-x\n",
    "    \n",
    "    apply_subtract = partial(subtract, x=perf_cl)\n",
    "    return apply_to_vals(perf_c, apply_subtract)\n",
    "\n",
    "def adjust_by_baseline(perf:dict, base:dict):\n",
    "    \n",
    "    perf = perf.copy()\n",
    "    for tfm in perf.keys():\n",
    "        perf[tfm] = perf[tfm]/base[tfm]\n",
    "        \n",
    "    return perf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# export\n",
    "def score_sam(df:pd.DataFrame, eval_func):\n",
    "    '''Compute performance score for SAM using 'eval_func'\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    df : pd.DataFrame\n",
    "        dataframe with columns \"nevus\" and \"melanoma\" which contain probability scores \n",
    "        and column \"ground_truth\" which contains a 0 (nevus) or 1 (melanoma)\n",
    "        \n",
    "    eval_func : callable\n",
    "        function which is used to evaluate the predictions \n",
    "        \n",
    "    Returns\n",
    "    -------\n",
    "    perf : dict\n",
    "        Stores performance score \n",
    "    '''\n",
    "    perf = dict()\n",
    "    preds = (torch.Tensor(np.array(df[[\"nevus\", \"melanoma\"]])), torch.LongTensor(np.array(df[\"ground_truth\"])))\n",
    "    perf[\"clean\"] = eval_func(preds)\n",
    "    \n",
    "    return perf\n",
    "\n",
    "def score_samc(df:pd.DataFrame, eval_func):\n",
    "    '''Compute performance score for SAM-C/SAM-C-Extra using 'eval_func'\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    df : pd.DataFrame\n",
    "        dataframe with columns \"nevus\" and \"melanoma\" which contain probability scores \n",
    "        and column \"ground_truth\" which contains a 0 (nevus) or 1 (melanoma)\n",
    "        \n",
    "    eval_func : callable\n",
    "        function which is used to evaluate the predictions \n",
    "        \n",
    "    Returns\n",
    "    -------\n",
    "    perf : dict\n",
    "        Stores performance score \n",
    "    '''\n",
    "    # add transformation type to df (required for scoring corruptions and perturbations)\n",
    "    df[\"tfms\"] = df.image_path.apply(lambda x: x.split(\"/\")[2])\n",
    "    \n",
    "    # add severity levels to df (only needed for corruptions)\n",
    "    df[\"severities\"] = df.image_path.apply(lambda x: x.split(\"/\")[3])\n",
    "    assert df[\"severities\"].nunique()==5, f\"Invalid number of severities ({df['severities'].nunique()})\"\n",
    "    \n",
    "    perf = dict()\n",
    "    for (k1, k2), group_df in df.groupby([\"tfms\", \"severities\"]):\n",
    "        preds = (torch.Tensor(np.array(group_df[[\"nevus\", \"melanoma\"]])), torch.LongTensor(np.array(group_df[\"ground_truth\"])))\n",
    "\n",
    "        if perf.get(k1): \n",
    "            perf[k1][k2] = eval_func(preds)\n",
    "        else:\n",
    "            perf[k1] = dict()\n",
    "            perf[k1][k2] = eval_func(preds)\n",
    "\n",
    "    perf = apply_to_vals(flatten_intra(perf), np.mean)\n",
    "    \n",
    "    return perf\n",
    "\n",
    "def score_samp(df:pd.DataFrame, eval_func):\n",
    "    '''Compute performance score for SAM-P using 'eval_func'\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    df : pd.DataFrame\n",
    "        dataframe with columns \"nevus\" and \"melanoma\" which contain probability scores \n",
    "        and column \"ground_truth\" which contains a 0 (nevus) or 1 (melanoma)\n",
    "        \n",
    "    eval_func : callable\n",
    "        function which is used to evaluate the predictions \n",
    "        \n",
    "    Returns\n",
    "    -------\n",
    "    perf : dict\n",
    "        Stores performance score \n",
    "    '''\n",
    "    # add transformation type to df (required for scoring corruptions and perturbations)\n",
    "    df[\"tfms\"] = df.image_path.apply(lambda x: x.split(\"/\")[2])\n",
    "    \n",
    "    perf = dict()\n",
    "    \n",
    "    for k1, group_df in df.groupby([\"tfms\"]):\n",
    "        preds = (torch.Tensor(np.array(group_df[[\"nevus\", \"melanoma\"]])), torch.LongTensor(np.array(group_df[\"ground_truth\"])))\n",
    "        perf[k1] = eval_func(k1, preds)\n",
    "        \n",
    "    return perf\n",
    "\n",
    "def save_perf(df:pd.DataFrame, out_path:str, out_file:str):\n",
    "    if not os.path.exists(out_path):\n",
    "        os.mkdir(out_path)\n",
    "    df.to_csv(f\"{out_path}/{out_file}.csv\", index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}