asr_lab1.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We start by importing the OpenFst Python wrapper:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import openfst_python as fst"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "partner = 's1608733'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then we create tables for our symbols"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "input_sym = fst.SymbolTable()\n",
    "output_sym = fst.SymbolTable()\n",
    "\n",
    "input_sym.add_symbol('<eps>') # by convention, <eps> always\n",
    "                              # has symbol zero\n",
    "input_sym.add_symbol('a')  # input symbols\n",
    "input_sym.add_symbol('b')\n",
    "\n",
    "output_sym.add_symbol('<eps>')  # output symbols\n",
    "output_sym.add_symbol('d')\n",
    "output_sym.add_symbol('c')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A **SymbolTable()** is simply a table associating symbols and indexes. We add symbols to the table with the method **add_symbol()**\n",
    "\n",
    "Now that we've got our symbol tables, we will build the FST itself:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "f = fst.Fst()\n",
    "\n",
    "f.set_input_symbols(input_sym)\n",
    "f.set_output_symbols(output_sym)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Our FST transduces the input to the outputs, so we set the symbol tables as such. Now, we want to add a number of states:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "s0 = f.add_state()\n",
    "s1 = f.add_state()\n",
    "s2 = f.add_state()\n",
    "s3 = f.add_state()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The output of the **add_state()** method is just the index assigned to the state, but it can be useful to assign that index to a variable to give it a more meaningful label.\n",
    "\n",
    "To create arcs between states, we do:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a = input_sym.find('a')\n",
    "b = input_sym.find('b')\n",
    "c = output_sym.find('c')\n",
    "d = output_sym.find('d')\n",
    "\n",
    "f.add_arc(s0, fst.Arc(a, c, None, s1))\n",
    "f.add_arc(s0, fst.Arc(b, d, None, s2))\n",
    "f.add_arc(s1, fst.Arc(a, c, None, s3))\n",
    "f.add_arc(s2, fst.Arc(b, d, None, s3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The syntax for the method **add_arc** is:\n",
    "\n",
    "**add_arc(*source state*, *arc to add*)**\n",
    "\n",
    "while the syntax for initializing a class **Arc()** is:\n",
    "\n",
    "**Arc(*input symbol index*, *output symbol index*, *weight*, *destination state*)**\n",
    "\n",
    "We use the **find()** method of the symbol tables to get the index of a certain label.\n",
    "\n",
    "Now we just add start and end states:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "f.set_start(s0)\n",
    "f.set_final(s3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And voila, our first FST, shown in the example above, is done!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercises"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def parse_lexicon(lex_file):\n",
    "    \"\"\"\n",
    "    Parse the lexicon file and return it in dictionary form.\n",
    "    \n",
    "    Args:\n",
    "        lex_file (str): filename of lexicon file with structure '<word> <phone1> <phone2>...'\n",
    "                        eg. peppers p eh p er z\n",
    "\n",
    "    Returns:\n",
    "        lex (dict): dictionary mapping words to list of phones\n",
    "    \"\"\"\n",
    "    \n",
    "    lex = {}  # create a dictionary for the lexicon entries (this could be a problem with larger lexica)\n",
    "    with open(lex_file, 'r') as f:\n",
    "        for line in f:\n",
    "            line = line.split()  # split at each space\n",
    "            lex[line[0]] = line[1:]  # first field the word, the rest is the phones\n",
    "    return lex"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "lex = parse_lexicon('lexicon.txt')\n",
    "lex"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "1. Write a function that generates symbol tables from a lexicon, using the provided phonetic dictionary in **lexicon.txt** and the helper function *parse_lexicon* in cells above. Use p_1, p_2, ..., eh_1, ... to denote the phone states. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_symbol_tables(lexicon, n=3):\n",
    "    '''\n",
    "    Return word, phone and state symbol tables based on the supplied lexicon\n",
    "    \n",
    "    Args:\n",
    "        lexicon (dict): lexicon to use, created from the parse_lexicon() function\n",
    "        n (int): number of states for each phone HMM\n",
    "        \n",
    "    Returns:\n",
    "        word_table (fst.SymbolTable): table of words\n",
    "        phone_table (fst.SymbolTable): table of phones\n",
    "        state_table (fst.SymbolTable): table of HMM phone-state IDs\n",
    "    '''\n",
    "        \n",
    "    state_table = fst.SymbolTable()\n",
    "    phone_table = fst.SymbolTable()\n",
    "    word_table = fst.SymbolTable()\n",
    "\n",
    "    state_table.add_symbol('𝜖')\n",
    "    phone_table.add_symbol('𝜖')\n",
    "    word_table.add_symbol('𝜖')\n",
    "    phones = set()\n",
    "    \n",
    "    for word, word_phones in sorted(lexicon.items(), key=lambda x: x[0]):\n",
    "        word_table.add_symbol(word)\n",
    "        print(word)\n",
    "        print(word_phones)\n",
    "        phones = phones.union({p for p in word_phones})\n",
    "\n",
    "    for phone in sorted(phones):\n",
    "        phone_table.add_symbol(phone)\n",
    "        for i in range(1, n + 1):\n",
    "            print(f\"{phone}_{i}\")\n",
    "            state_table.add_symbol(f\"{phone}_{i}\")\n",
    "\n",
    "    return word_table, phone_table, state_table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "word_table, phone_table, state_table = generate_symbol_tables(lex)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "2. Create functions generating WFSTs representing the following HMMs:\n",
    "\n",
    "    1. A 3-state left-to-right phone HMM with self-loops\n",
    "    2. A parallel-path left-to-right model (shown on Slide 10 of Lecture 4)\n",
    "    3. An ergodic HMM with $n$ states (you will need to think about how to handle final states)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_phone_wfst(f, start_state, phone, n):\n",
    "    \"\"\"\n",
    "    Generate a WFST representing an n-state left-to-right phone HMM.\n",
    "    \n",
    "    Args:\n",
    "        f (fst.Fst()): an FST object, assumed to exist already\n",
    "        start_state (int): the index of the first state, assumed to exist already\n",
    "        phone (str): the phone label \n",
    "        n (int): number of states of the HMM excluding start and end\n",
    "        \n",
    "    Returns:\n",
    "        the final state of the FST\n",
    "    \"\"\"\n",
    "    \n",
    "    current_state = start_state\n",
    "    \n",
    "    for i in range(1, n+1):\n",
    "\n",
    "        in_label = state_table.find('{}_{}'.format(phone, i))\n",
    "        out_label = state_table.find('𝜖')\n",
    "        \n",
    "        # special case for last state\n",
    "        if i == n:\n",
    "            f.add_arc(current_state, fst.Arc(in_label, out_label, None, current_state))\n",
    "            continue\n",
    "        \n",
    "        next_state = f.add_state()\n",
    "        \n",
    "        f.add_arc(current_state,fst.Arc(in_label, out_label, None, current_state))\n",
    "        f.add_arc(current_state,fst.Arc(in_label, out_label, None, next_state))\n",
    "        current_state = next_state\n",
    "        \n",
    "    final_state = f.add_state()\n",
    "    f.add_arc(current_state, fst.Arc(in_label, phone_table.find(phone), None, final_state))\n",
    "    current_state = final_state\n",
    "\n",
    "    return current_state\n",
    "\n",
    "f = fst.Fst()\n",
    "start = f.add_state()\n",
    "f.set_start(start)\n",
    "\n",
    "last_state = generate_phone_wfst(f, start, 'p', 3)\n",
    "\n",
    "f.set_input_symbols(state_table)\n",
    "f.set_output_symbols(phone_table)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_parallel_path_wfst(f, start_state, n):\n",
    "    \"\"\"\n",
    "    Generate a WFST representing an n-state parallel-path left-to-right HMM\n",
    "    \n",
    "    Args:\n",
    "        f (fst.Fst()): an FST object, assumed to exist already\n",
    "        start_state (int): the index of the first state, assumed to exist already\n",
    "        n (int): number of states of the HMM excluding start and end\n",
    "        \n",
    "    Returns:\n",
    "        the final state of the FST\n",
    "    \"\"\"\n",
    "    \n",
    "    current_state = start_state\n",
    "    next_state = start_state\n",
    "    \n",
    "    for i in range(n):\n",
    "    \n",
    "        in_label = state_table.find('𝜖')\n",
    "        out_label = state_table.find('𝜖')\n",
    "\n",
    "        if i < n - 1:\n",
    "            if i == 0:\n",
    "                next_state = f.add_state()\n",
    "                f.add_arc(current_state, fst.Arc(in_label, out_label, None, current_state))\n",
    "                f.add_arc(current_state, fst.Arc(in_label, out_label, None, next_state))\n",
    "                f.add_arc(next_state, fst.Arc(in_label, out_label, None, next_state))\n",
    "        \n",
    "            next_next_state = f.add_state()\n",
    "            f.add_arc(current_state, fst.Arc(in_label, out_label, None, next_next_state))\n",
    "            f.add_arc(next_state, fst.Arc(in_label, out_label, None, next_next_state))\n",
    "            \n",
    "            if i < n - 2:\n",
    "                f.add_arc(next_next_state, fst.Arc(in_label, out_label, None, next_next_state))\n",
    "        \n",
    "        current_state = next_state\n",
    "        next_state = next_next_state\n",
    "\n",
    "    f.set_final(current_state)\n",
    "    return current_state\n",
    "\n",
    "f = fst.Fst()\n",
    "start = f.add_state()\n",
    "f.set_start(start)\n",
    "\n",
    "last_state = generate_parallel_path_wfst(f, start, 4)\n",
    "\n",
    "f.set_input_symbols(state_table)\n",
    "f.set_output_symbols(phone_table)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_ergodic_wfst(f, start_state, n):\n",
    "    \"\"\"\n",
    "    Generate a WFST representing an n-state ergodic HMM\n",
    "    \n",
    "    Args:\n",
    "        f (fst.Fst()): an FST object, assumed to exist already\n",
    "        start_state (int): the index of the first state, assumed to exist already\n",
    "        n (int): number of states of the HMM excluding start and end\n",
    "        \n",
    "    Returns:\n",
    "        the final state of the FST\n",
    "    \"\"\"    \n",
    "    \n",
    "    current_state = start_state\n",
    "    \n",
    "    for i in range(n+1):\n",
    "        \n",
    "        in_label = state_table.find('𝜖')\n",
    "        out_label = state_table.find('𝜖')\n",
    "        \n",
    "        if i == 0:\n",
    "            for _ in range(1, n+1):\n",
    "                f.add_arc(current_state, fst.Arc(in_label, out_label, None, f.add_state()))\n",
    "                \n",
    "        else:\n",
    "            for j in range(1, n+1):\n",
    "                f.add_arc(current_state, fst.Arc(in_label, out_label, None, j))\n",
    "\n",
    "        current_state = i+1\n",
    "\n",
    "    final_state = f.add_state()\n",
    "    f.set_final(final_state)\n",
    "    for i in range(1, n+1):\n",
    "        f.add_arc(i, fst.Arc(in_label, out_label, None, final_state))\n",
    "    current_state = final_state\n",
    "    return current_state\n",
    "\n",
    "f = fst.Fst()\n",
    "start = f.add_state()\n",
    "f.set_start(start)\n",
    "\n",
    "last_state = generate_ergodic_wfst(f, start, 5)\n",
    "\n",
    "f.set_input_symbols(state_table)\n",
    "f.set_output_symbols(phone_table)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "3. Write a function to generate an HMM for any word in the lexicon using $n$ states per phone by concatenating $n$-state phone HMMs. Make use of your *generate_phone_wfst()* function. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_word_wfst(f, start_state, word, n):\n",
    "    \"\"\" Generate a WFST for any word in the lexicon, composed of n-state phone WFSTs.\n",
    "        This will currently output word labels.  \n",
    "        Exercise: could you modify this function and the one above to output a single \n",
    "        phone label instead?\n",
    "    \n",
    "    Args:\n",
    "        f (fst.Fst()): an FST object, assumed to exist already\n",
    "        start_state (int): the index of the first state, assumed to exist already\n",
    "        word (str): the word to generate\n",
    "        n (int): states per phone HMM\n",
    "        \n",
    "    Returns:\n",
    "        the constructed WFST\n",
    "    \n",
    "    \"\"\"\n",
    "\n",
    "    current_state = start_state\n",
    "    \n",
    "    for phone in lex[word]:\n",
    "        last_state = generate_phone_wfst(f, current_state, phone, n)\n",
    "        current_state = last_state\n",
    "    \n",
    "    return f\n",
    "\n",
    "f = fst.Fst()\n",
    "start = f.add_state()\n",
    "f.set_start(start)\n",
    "\n",
    "generate_word_wfst(f, start, 'peppers', 3)\n",
    "f.set_input_symbols(state_table)\n",
    "f.set_output_symbols(phone_table)\n",
    "\n",
    "from subprocess import check_call\n",
    "f.draw('tmp.dot', portrait=True)\n",
    "check_call(['dot', '-Tpng', '-Gdpi=200', 'tmp.dot', '-o', 'tmp.png'])\n",
    "from IPython.display import Image\n",
    "Image(filename='tmp.png')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "4. Generate an HMM that can recognise:\n",
    "\n",
    "    1. any phone contained in the lexicon\n",
    "    2. any sequence of phones contained in the lexicon\n",
    "    \n",
    "Think of the difference between the first and the second HMM. Again, previous functions should help in the task. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_phone_recognition_wfst(n):\n",
    "    \"\"\" generate a HMM to recognise any single phone in the lexicon\n",
    "    \n",
    "    Args:\n",
    "        n (int): states per phone HMM\n",
    "\n",
    "    Returns:\n",
    "        the constructed WFST\n",
    "    \n",
    "    \"\"\"\n",
    "    \n",
    "    f = fst.Fst()\n",
    "    \n",
    "    # create a single start state\n",
    "    start_state = f.add_state()\n",
    "    f.set_start(start_state)\n",
    "    # create a single end state\n",
    "    end_state = f.add_state()\n",
    "    f.set_final(end_state)\n",
    "    \n",
    "    # collect all unique phones from the lexicon\n",
    "    phones = sorted({phone for word, phones in lex.items() for phone in phones})\n",
    "\n",
    "    for phone in phones:\n",
    "        phone_start_state = f.add_state()\n",
    "        phone_end_state = generate_phone_wfst(f, phone_start_state, phone, n)\n",
    "        # transition into phone HMM from global start state\n",
    "        f.add_arc(start_state, fst.Arc(state_table.find('𝜖'), state_table.find('𝜖'), None, phone_start_state))\n",
    "        # transition from phone HMM into global end state\n",
    "        f.add_arc(phone_end_state, fst.Arc(state_table.find('𝜖'), state_table.find('𝜖'), None, end_state))\n",
    "\n",
    "    return f\n",
    "\n",
    "f = generate_phone_recognition_wfst(3)\n",
    "f.set_input_symbols(state_table)\n",
    "f.set_output_symbols(phone_table)\n",
    "\n",
    "from subprocess import check_call\n",
    "f.draw('tmp.dot', portrait=True)\n",
    "check_call(['dot', '-Tpng', '-Gdpi=300', 'tmp.dot', '-o', 'tmp.png'])\n",
    "from IPython.display import Image\n",
    "Image(filename='tmp.png')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_phone_sequence_recognition_wfst(n):\n",
    "    \"\"\" generate a HMM to recognise any single phone sequence in the lexicon\n",
    "    \n",
    "    Args:\n",
    "        n (int): states per phone HMM\n",
    "\n",
    "    Returns:\n",
    "        the constructed WFST\n",
    "    \n",
    "    \"\"\"\n",
    "    \n",
    "    f = fst.Fst()\n",
    "    \n",
    "    # create a single start state\n",
    "    start_state = f.add_state()\n",
    "    f.set_start(start_state)\n",
    "    # create a single end state\n",
    "    end_state = f.add_state()\n",
    "    f.set_final(end_state)\n",
    "    \n",
    "    # collect all unique phones from the lexicon\n",
    "    phones = sorted({phone for word, phones in lex.items() for phone in phones})\n",
    "\n",
    "    for phone in phones:\n",
    "        phone_start_state = f.add_state()\n",
    "        phone_end_state = generate_phone_wfst(f, phone_start_state, phone, n)\n",
    "        # transition into phone HMM from global start state\n",
    "        f.add_arc(start_state, fst.Arc(state_table.find('𝜖'), state_table.find('𝜖'), None, phone_start_state))\n",
    "        # transition from phone HMM to gloabl end state\n",
    "        f.add_arc(phone_end_state, fst.Arc(state_table.find('𝜖'), state_table.find('𝜖'), None, end_state))\n",
    "    \n",
    "    # loop back to start state to recognize more phones\n",
    "    f.add_arc(end_state, fst.Arc(state_table.find('𝜖'), state_table.find('𝜖'), None, start_state))\n",
    "\n",
    "    return f\n",
    "\n",
    "f = generate_phone_sequence_recognition_wfst(3)\n",
    "f.set_input_symbols(state_table)\n",
    "f.set_output_symbols(phone_table)\n",
    "\n",
    "from subprocess import check_call\n",
    "f.draw('tmp.dot', portrait=True)\n",
    "check_call(['dot', '-Tpng', '-Gdpi=300', 'tmp.dot', '-o', 'tmp.png'])\n",
    "from IPython.display import Image\n",
    "Image(filename='tmp.png')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "5. You have decided that the best way to start teaching a friend English is to have them learn the notorious tongue-twister *peter piper picked a peck of pickled peppers*.  Write a function that creates an HMM that can recognize any sequence of these words."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_word_sequence_recognition_wfst(n):\n",
    "    \"\"\" generate a HMM to recognise any single word sequence for words in the lexicon\n",
    "    \n",
    "    Args:\n",
    "        n (int): states per phone HMM\n",
    "\n",
    "    Returns:\n",
    "        the constructed WFST\n",
    "    \n",
    "    \"\"\"\n",
    "    \n",
    "    f = fst.Fst()\n",
    "    \n",
    "    # create a single start state\n",
    "    start_state = f.add_state()\n",
    "    f.set_start(start_state)\n",
    "    # create a single end state\n",
    "    end_state = f.add_state()\n",
    "    f.set_final(end_state)\n",
    "    \n",
    "    for word in lex.keys():\n",
    "        current_state = f.add_state()\n",
    "        f.add_arc(start_state, fst.Arc(state_table.find('𝜖'), state_table.find('𝜖'), None, current_state))\n",
    "    \n",
    "        for phone in lex[word]:\n",
    "            current_state = generate_phone_wfst(f, current_state, phone, n)\n",
    "        \n",
    "        # transition from word HMM into global end state\n",
    "        f.add_arc(current_state, fst.Arc(state_table.find('𝜖'), state_table.find('𝜖'), None, end_state))\n",
    "        \n",
    "    # loop back to start state to recognize more words\n",
    "    f.add_arc(end_state, fst.Arc(state_table.find('𝜖'), state_table.find('𝜖'), None, start_state))\n",
    "        \n",
    "    return f\n",
    "\n",
    "f = generate_word_sequence_recognition_wfst(3)\n",
    "f.set_input_symbols(state_table)\n",
    "f.set_output_symbols(phone_table)\n",
    "\n",
    "from subprocess import check_call\n",
    "f.draw('tmp.dot', portrait=True)\n",
    "check_call(['dot', '-Tpng', '-Gdpi=300', 'tmp.dot', '-o', 'tmp.png'])\n",
    "from IPython.display import Image\n",
    "Image(filename='tmp.png')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# If you have more time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Probabilities in WFSTs are traditionally expressed in negative log format, that is, the weight $w$ on an arc transitioning between states $i$ and $j$ is given by $w=-\\log a_{ij}$, where $a_{ij}$ is the HMM transition probability.\n",
    "\n",
    "6. Add weights to your WFSTs corresponding to transition probabilities.  Assume that the probability of a self-loop is $0.1$, and that when transitioning *between* separate phones or words, the probabilities are uniform over all transitions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.7.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}