build-your-own-image-classifier3.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#hide\n",
    "from fastai.vision.all import *\n",
    "from ipywidgets import widgets \n",
    "from fastai.vision.widgets import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Cell\n",
    "from fastai.callback.preds import MCDropoutCallback\n",
    "from fastai.learner import Learner\n",
    "from fastcore.foundation import patch, L\n",
    "from fastcore.basics import tuplify,detuplify\n",
    "from fastai.torch_core import to_np\n",
    "from fastai.data.transforms import get_image_files\n",
    "from fastai.vision.core import PILImage\n",
    "\n",
    "# Cell\n",
    "from collections import Counter\n",
    "import seaborn as sns\n",
    "import torch\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "\n",
    "# Cell\n",
    "def entropy(probs):\n",
    "    \"\"\"Return the prediction of a T*N*C tensor with :\n",
    "        - T : the number of samples\n",
    "        - N : the batch size\n",
    "        - C : the number of classes\n",
    "    \"\"\"\n",
    "    mean_probs = probs.mean(dim=0)\n",
    "    entrop = - (torch.log(mean_probs) * mean_probs).sum(dim=1)\n",
    "    return entrop\n",
    "\n",
    "def uncertainty_best_probability(probs):\n",
    "    \"\"\"Return the standard deviation of the most probable class\"\"\"\n",
    "    idx = probs.mean(dim=0).argmax(dim=1)\n",
    "\n",
    "    std = probs[:, torch.arange(len(idx)), idx].std(dim=0)\n",
    "\n",
    "    return std\n",
    "\n",
    "def BALD(probs):\n",
    "    \"\"\"Information Gain, distance between the entropy of averages and average of entropy\"\"\"\n",
    "    entrop1 = entropy(probs)\n",
    "    entrop2 = - (torch.log(probs) * probs).sum(dim=2)\n",
    "    entrop2 = entrop2.mean(dim=0)\n",
    "\n",
    "    ig = entrop1 - entrop2\n",
    "    return ig\n",
    "\n",
    "def top_k_uncertainty(s, k=5, reverse=True):\n",
    "    \"\"\"Return the top k indexes\"\"\"\n",
    "    sorted_s = sorted(list(zip(torch.arange(len(s)), s)),\n",
    "                      key=lambda x: x[1], reverse=reverse)\n",
    "    output = [sorted_s[i][0] for i in range(k)]\n",
    "\n",
    "def plot_hist_groups(pred,y,metric,bins=None,figsize=(16,16)):\n",
    "    TP = to_np((pred.mean(dim=0).argmax(dim=1) == y) & (y == 1))\n",
    "    TN = to_np((pred.mean(dim=0).argmax(dim=1) == y) & (y == 0))\n",
    "    FP = to_np((pred.mean(dim=0).argmax(dim=1) != y) & (y == 0))\n",
    "    FN = to_np((pred.mean(dim=0).argmax(dim=1) != y) & (y == 1))\n",
    "\n",
    "    result = metric(pred)\n",
    "\n",
    "    TP_result = result[TP]\n",
    "    TN_result = result[TN]\n",
    "    FP_result = result[FP]\n",
    "    FN_result = result[FN]\n",
    "\n",
    "    fig,ax = plt.subplots(2,2,figsize=figsize)\n",
    "\n",
    "    sns.distplot(TP_result,ax=ax[0,0],bins=bins)\n",
    "    ax[0,0].set_title(f\"True positive\")\n",
    "\n",
    "    sns.distplot(TN_result,ax=ax[0,1],bins=bins)\n",
    "    ax[0,1].set_title(f\"True negative\")\n",
    "\n",
    "    sns.distplot(FP_result,ax=ax[1,0],bins=bins)\n",
    "    ax[1,0].set_title(f\"False positive\")\n",
    "\n",
    "    sns.distplot(FN_result,ax=ax[1,1],bins=bins)\n",
    "    ax[1,1].set_title(f\"False negative\")\n",
    "    return output\n",
    "\n",
    "# Cell\n",
    "@patch\n",
    "def bayes_get_preds(self:Learner, ds_idx=1, dl=None, n_sample=10,\n",
    "                    act=None,with_loss=False, **kwargs):\n",
    "    \"\"\"Get MC Dropout predictions from a learner, and eventually reduce the samples\"\"\"\n",
    "    cbs = [MCDropoutCallback()]\n",
    "    if 'cbs' in kwargs:\n",
    "        kw_cbs = kwargs.pop('cbs')\n",
    "        if 'MCDropoutCallback' not in L(kw_cbs).attrgot('name'):\n",
    "            cbs = kw_cbs + cbs\n",
    "    preds = []\n",
    "    with self.no_bar():\n",
    "        for i in range(n_sample):\n",
    "            pred, y = self.get_preds(ds_idx=ds_idx,dl=dl,act=act,\n",
    "                                     with_loss=with_loss, cbs=cbs, **kwargs)\n",
    "            # pred = n_dl x n_vocab\n",
    "            preds.append(pred)\n",
    "    preds = torch.stack(preds)\n",
    "    ents = entropy(preds)\n",
    "    mean_preds = preds.mean(dim=0)\n",
    "    max_preds = mean_preds.max(dim=1)\n",
    "    best_guess = max_preds.indices\n",
    "    best_prob = max_preds.values\n",
    "    best_cat = L(best_guess,use_list=True).map(lambda o: self.dls.vocab[o.item()])\n",
    "    return preds, mean_preds, ents,best_guess, best_prob, best_cat\n",
    "\n",
    "# Cell\n",
    "@patch\n",
    "def bayes_predict(self:Learner,item, rm_type_tfms=None, with_input=False,\n",
    "                  sample_size=10,reduce=True):\n",
    "    \"gets a sample distribution of predictions and computes entropy\"\n",
    "    dl = self.dls.test_dl([item], rm_type_tfms=rm_type_tfms, num_workers=0)\n",
    "\n",
    "    # modify get_preds to get distributed samples\n",
    "    collect_preds = []\n",
    "    collect_targs = []\n",
    "    collect_dec_preds = []\n",
    "    collect_inp = None\n",
    "    cbs = [MCDropoutCallback()]\n",
    "    with self.no_bar():\n",
    "        for j in range(sample_size):\n",
    "            inp,preds,_,dec_preds = self.get_preds(dl=dl, with_input=True,\n",
    "                                                   with_decoded=True,\n",
    "                                                   cbs=cbs)\n",
    "            i = getattr(self.dls, 'n_inp', -1)\n",
    "            inp = (inp,) if i==1 else tuplify(inp)\n",
    "            dec = self.dls.decode_batch(inp + tuplify(dec_preds))[0]\n",
    "            dec_inp,dec_targ = map(detuplify, [dec[:i],dec[i:]])\n",
    "            # res = dec_targ,dec_preds[0],preds[0]\n",
    "            if with_input and collect_inp is None: # collect inp first iter only\n",
    "                   collect_inp = dec_inp\n",
    "            collect_targs.append(dec_targ)\n",
    "            collect_dec_preds.append(dec_preds[0])\n",
    "            collect_preds.append(preds[0])\n",
    "    dist_preds = torch.stack(collect_preds)\n",
    "    dist_dec_preds = L(collect_dec_preds).map(lambda o: o.item())\n",
    "    dist_targs = L(collect_targs)\n",
    "    res1 = (dist_targs, dist_dec_preds, dist_preds)\n",
    "\n",
    "    mean_pred = dist_preds.mean(dim=0)\n",
    "    ent = entropy(dist_preds.unsqueeze(1)).item()\n",
    "    best_guess = torch.argmax(mean_pred).item()\n",
    "    best_prob = mean_pred[best_guess].item()\n",
    "    best_cat = self.dls.vocab[best_guess]\n",
    "    res2 = (ent, best_prob, best_guess, best_cat)\n",
    "\n",
    "    if reduce:\n",
    "        if len(dist_targs.unique()) > 1:\n",
    "            targ = Counter(dist_targs)\n",
    "        else:\n",
    "            targ = dist_targs.unique()[0]\n",
    "\n",
    "        if len(dist_dec_preds.unique()) > 1:\n",
    "            dec_pred = Counter(dist_dec_preds)\n",
    "        else:\n",
    "            dec_pred = dist_dec_preds.unique()[0]\n",
    "        res1 = (targ, dec_pred, mean_pred)\n",
    "\n",
    "    res = res1 + res2\n",
    "    if with_input:\n",
    "        res = (collect_inp,) + res\n",
    "    return res\n",
    "\n",
    "\n",
    "# Cell\n",
    "@patch\n",
    "def bayes_predict_with_uncertainty(self:Learner, item, rm_type_tfms=None, with_input=False, threshold_entropy=0.2, sample_size=10, reduce=True):\n",
    "    \"gets prediction results plus if prediction passes entropy threshold\"\n",
    "    res = self.bayes_predict(item,rm_type_tfms=rm_type_tfms,\n",
    "                             with_input=with_input, sample_size=sample_size,\n",
    "                             reduce=reduce)\n",
    "    ent = res[4] if with_input else res[3]\n",
    "    return (ent < threshold_entropy,) + res\n",
    "\n",
    "# Cell\n",
    "@patch\n",
    "def bayes_build_inference_dfdlpreds(self:Learner, path, dataset, item_count=100,n_sample=10):\n",
    "    items = get_image_files(path).shuffle()[:item_count]\n",
    "    dl = self.dls.test_dl(items.map(lambda o: PILImage.create(o)), num_workers=0)\n",
    "    res = self.bayes_get_preds(dl=dl,n_sample=n_sample)\n",
    "    ents = res[2]\n",
    "    preds = res[0]\n",
    "    unc = uncertainty_best_probability(preds)\n",
    "    bald = BALD(preds)\n",
    "    df = pd.DataFrame(pd.Series(items,name='image_files'))\n",
    "    df['entropy'] = pd.Series(ents,name='entropy')\n",
    "    df['best_prob_uncertainty'] = pd.Series(unc,name='best_prob_uncertainty')\n",
    "    df['bald'] = pd.Series(bald,name='bald')\n",
    "    df['dataset'] = dataset\n",
    "    return (df,dl, preds)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#hide\n",
    "path = Path()\n",
    "# path.ls(file_exts='.pkl')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "learn_inf = load_learner(path/'export.pkl')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# My Very Own (Really Cool!)  Image Classifier with a Fail (Uncertainty) Detector"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Classifier Categories:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for category in learn_inf.dls.vocab:\n",
    "    print(category)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def singularize(word):\n",
    "    prefix = 'a '\n",
    "    if word[-1] == 's':\n",
    "        word = word[:-1]\n",
    "    if word[0] in ('a','e','i','o','u'):\n",
    "        prefix = 'an '\n",
    "    return prefix + word\n",
    "\n",
    "def handle_bayes_prediction_with_uncertainty(res):\n",
    "    pass_threshold = res[0]\n",
    "    best_cat = singularize(res[7])\n",
    "    best_prob = res[5] * 100.\n",
    "    best_unc = res[4]\n",
    "    unc_msg1 = f\"Doesn't look like anything to me, but it could be {best_cat}\"\n",
    "    unc_msg2 = f' with a probability of {best_prob:.2f} percent ' \n",
    "    unc_msg3 = f'and uncertainty(entropy) of {best_unc:.2f}'\n",
    "    uncertain_msg = unc_msg1 + unc_msg2 + unc_msg3\n",
    "    certain_msg = f'To me, that looks like {best_cat}, with a probability of {best_prob:.2f}'\n",
    "    if not pass_threshold: \n",
    "        return uncertain_msg\n",
    "    return certain_msg\n",
    "\n",
    "btn_upload = SimpleNamespace(data = ['images/purple_dog.jpg'])\n",
    "img = PILImage.create(btn_upload.data[-1])\n",
    "out_pl = widgets.Output()\n",
    "out_pl.clear_output()\n",
    "with out_pl: display(img.to_thumb(200,200))\n",
    "res = learn_inf.bayes_predict_with_uncertainty(img)\n",
    "\n",
    "lbl_pred = widgets.Label()\n",
    "lbl_pred.value = handle_bayes_prediction_with_uncertainty(res)\n",
    "btn_run = widgets.Button(description='Classify')\n",
    "\n",
    "def on_click_classify(change):\n",
    "    img = PILImage.create(btn_upload.data[-1])\n",
    "    out_pl.clear_output()\n",
    "    with out_pl: display(img.to_thumb(200,200))\n",
    "    res = learn_inf.bayes_predict_with_uncertainty(img)\n",
    "    lbl_pred.value = handle_bayes_prediction_with_uncertainty(res)\n",
    "\n",
    "btn_run.on_click(on_click_classify)\n",
    "#Putting back btn_upload to a widget for next cell\n",
    "btn_upload = widgets.FileUpload()\n",
    "#hide_output\n",
    "VBox([widgets.Label('UPLOAD AN IMAGE AND CLASSIFY!'), \n",
    "      btn_upload, btn_run, out_pl, lbl_pred])"
   ]
  }
 ],
 "metadata": {
  "jupytext": {
   "split_at_heading": true
  },
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.7"
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}