405_levenshtein_distance_wer_cer.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Levenshtein distance and WER/CER\n",
    "\n",
    "(References)\n",
    "\n",
    "- https://www.rev.ai/blog/how-to-calculate-word-error-rate/\n",
    "- https://en.wikipedia.org/wiki/Levenshtein_distance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import typing as t\n",
    "from collections import deque\n",
    "\n",
    "import numpy as np\n",
    "\n",
    "\n",
    "def levenshtein_distance(s: t.Sequence, t: t.Sequence):\n",
    "    # for all i and j, d[i,j] will hold the Levenshtein distance between\n",
    "    # the first i characters of s and the first j characters of t\n",
    "\n",
    "    m = len(s) + 1\n",
    "    n = len(t) + 1\n",
    "\n",
    "    d = np.zeros((m, n))\n",
    "\n",
    "    # source prefixes can be transformed into empty string by\n",
    "    # dropping all characters\n",
    "    for i in range(1, m):\n",
    "        d[i, 0] = i\n",
    "\n",
    "    # target prefixes can be reached from empty source prefix\n",
    "    # by inserting every character\n",
    "    for j in range(1, n):\n",
    "        d[0, j] = j\n",
    "\n",
    "    for j in range(1, n):\n",
    "        for i in range(1, m):\n",
    "            if s[i - 1] == t[j - 1]:\n",
    "                substitution_cost = 0\n",
    "            else:\n",
    "                substitution_cost = 1\n",
    "\n",
    "            d[i, j] = min(\n",
    "                d[i - 1, j] + 1,  # deletion\n",
    "                d[i, j - 1] + 1,  # insertion\n",
    "                d[i - 1, j - 1] + substitution_cost,  # substitution\n",
    "            )\n",
    "\n",
    "    return d[m - 1, n - 1], d\n",
    "\n",
    "\n",
    "def levenshtein_distance_details(s1, s2, d, separator=\" \"):\n",
    "    i = d.shape[0] - 1\n",
    "    j = d.shape[1] - 1\n",
    "\n",
    "    result = deque()\n",
    "    prev_label = None\n",
    "    prev_tokens_same = deque()\n",
    "    prev_tokens_deleted = deque()\n",
    "    prev_tokens_inserted = deque()\n",
    "\n",
    "    SAME = \"SAME\"\n",
    "    DELETED = \"DEL\"\n",
    "    INSERTED = \"INS\"\n",
    "    SUBSTITUTED = \"SUBS\"\n",
    "\n",
    "    def prepend_prev_label():\n",
    "        if prev_label == SAME:\n",
    "            result.appendleft((prev_label, separator.join(prev_tokens_same)))\n",
    "            prev_tokens_same.clear()\n",
    "        elif prev_label == SUBSTITUTED:\n",
    "            result.appendleft(\n",
    "                (\n",
    "                    prev_label,\n",
    "                    separator.join(prev_tokens_deleted),\n",
    "                    separator.join(prev_tokens_inserted),\n",
    "                )\n",
    "            )\n",
    "            prev_tokens_deleted.clear()\n",
    "            prev_tokens_inserted.clear()\n",
    "        elif prev_label == DELETED:\n",
    "            result.appendleft((prev_label, separator.join(prev_tokens_deleted)))\n",
    "            prev_tokens_deleted.clear()\n",
    "        elif prev_label == INSERTED:\n",
    "            result.appendleft((prev_label, separator.join(prev_tokens_inserted)))\n",
    "            prev_tokens_inserted.clear()\n",
    "\n",
    "    def check_and_process_label(current_label):\n",
    "        if prev_label is None:\n",
    "            return\n",
    "\n",
    "        if prev_label != current_label:\n",
    "            prepend_prev_label()\n",
    "\n",
    "    while i > 0 or j > 0:\n",
    "        v_current = d[i, j]\n",
    "        v_deletion = d[i - 1, j]\n",
    "        v_insertion = d[i, j - 1]\n",
    "        v_diagonal = d[i - 1, j - 1]\n",
    "\n",
    "        v_min = min(v_deletion, v_insertion, v_diagonal)\n",
    "\n",
    "        if v_min == v_current:\n",
    "            check_and_process_label(SAME)\n",
    "            prev_tokens_same.appendleft(s1[i - 1])\n",
    "            prev_label = SAME\n",
    "            i -= 1\n",
    "            j -= 1\n",
    "        elif v_min == v_diagonal:\n",
    "            check_and_process_label(SUBSTITUTED)\n",
    "            prev_tokens_deleted.appendleft(s1[i - 1])\n",
    "            prev_tokens_inserted.appendleft(s2[j - 1])\n",
    "            prev_label = SUBSTITUTED\n",
    "            i -= 1\n",
    "            j -= 1\n",
    "        elif v_min == v_deletion:\n",
    "            check_and_process_label(DELETED)\n",
    "            prev_tokens_deleted.appendleft(s1[i - 1])\n",
    "            prev_label = DELETED\n",
    "            i -= 1\n",
    "        elif v_min == v_insertion:\n",
    "            check_and_process_label(INSERTED)\n",
    "            prev_tokens_inserted.appendleft(s2[j - 1])\n",
    "            prev_label = INSERTED\n",
    "            j -= 1\n",
    "        else:\n",
    "            raise ValueError(\"Found a case not expected within d\")\n",
    "\n",
    "    prepend_prev_label()\n",
    "\n",
    "    return result\n",
    "\n",
    "def WER(ref_sentence: str, hyp_sentence: str, requires_details: bool = False):\n",
    "    sep = \" \"\n",
    "    tokenize = lambda s: s.split(sep)\n",
    "    ref_tokens = tokenize(ref_sentence)\n",
    "    hyp_tokens = tokenize(hyp_sentence)\n",
    "    val, d = levenshtein_distance(ref_tokens, hyp_tokens)\n",
    "    return (\n",
    "        val / len(ref_tokens),\n",
    "        levenshtein_distance_details(ref_tokens, hyp_tokens, d, separator=sep)\n",
    "        if requires_details\n",
    "        else None,\n",
    "    )\n",
    "\n",
    "def CER(ref_word: str, hyp_word: str, requires_details: bool = False):\n",
    "    tokenize = lambda s: s.replace(\" \", \"\")\n",
    "    ref_tokens = tokenize(ref_word)\n",
    "    hyp_tokens = tokenize(hyp_word)\n",
    "    val, d = levenshtein_distance(ref_tokens, hyp_tokens)\n",
    "    return (\n",
    "        val / len(ref_tokens),\n",
    "        levenshtein_distance_details(ref_tokens, hyp_tokens, d, separator=\"\")\n",
    "        if requires_details\n",
    "        else None,\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "WER(s1, s2, requires_details=True)=(0.3793103448275862, deque([('SAME', 'We wanted people to know that'), ('INS', 'how'), ('SUBS', \"we've got something brand new\", 'to me where i know'), ('SAME', 'and essentially this product is'), ('DEL', 'uh'), ('SAME', 'what we call'), ('SUBS', 'disruptive', 'scripted'), ('SAME', 'changes the way'), ('DEL', 'that'), ('SAME', 'people'), ('SUBS', 'interact with', 'are rapid'), ('SAME', 'technology')]))\n",
      "\n",
      "CER(s1, s2, requires_details=True)=(0.26573426573426573, deque([('SAME', 'Wewantedpeopletoknowthat'), ('DEL', \"we've\"), ('SUBS', 'g', 'h'), ('SAME', 'o'), ('SUBS', 'ts', 'wt'), ('SAME', 'ome'), ('SUBS', 't', 'w'), ('SAME', 'h'), ('DEL', 'ing'), ('SUBS', 'b', 'e'), ('SAME', 'r'), ('SUBS', 'and', 'eik'), ('SAME', 'n'), ('SUBS', 'e', 'o'), ('SAME', 'wandessentiallythisproductis'), ('DEL', 'uh'), ('SAME', 'whatwecall'), ('DEL', 'd'), ('SUBS', 'is', 'sc'), ('SAME', 'r'), ('SUBS', 'u', 'i'), ('SAME', 'pt'), ('DEL', 'i'), ('SUBS', 've', 'ed'), ('SAME', 'changestheway'), ('DEL', 'that'), ('SAME', 'people'), ('DEL', 'i'), ('SUBS', 'nt', 'ar'), ('SAME', 'era'), ('DEL', 'ct'), ('SUBS', 'w', 'p'), ('SAME', 'i'), ('DEL', 't'), ('SUBS', 'h', 'd'), ('SAME', 'technology')]))\n"
     ]
    }
   ],
   "source": [
    "s1 = \"We wanted people to know that we've got something brand new and essentially this product is uh what we call disruptive changes the way that people interact with technology\"\n",
    "s2 = \"We wanted people to know that how to me where i know and essentially this product is what we call scripted changes the way people are rapid technology\"\n",
    "\n",
    "print(f\"{WER(s1, s2, requires_details=True)=}\\n\")\n",
    "print(f\"{CER(s1, s2, requires_details=True)=}\")"
   ]
  }
 ],
 "metadata": {
  "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.8.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}