CYK_parser.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 112,
   "metadata": {},
   "outputs": [],
   "source": [
    "#-*- coding: utf-8 -*-\n",
    "\n",
    "from __future__ import print_function, unicode_literals\n",
    "import os\n",
    "import sys\n",
    "sys.path.append(MELT_DIR)\n",
    "import codecs\n",
    "import random\n",
    "import pickle\n",
    "from time import time\n",
    "import numpy as np\n",
    "import string\n",
    "import re\n",
    "import nltk\n",
    "from nltk import Tree\n",
    "from collections import defaultdict\n",
    "from nltk.corpus.reader.bracket_parse import BracketParseCorpusReader"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 149,
   "metadata": {},
   "outputs": [],
   "source": [
    "class CYK_parser(object):\n",
    "    def __init__(self):\n",
    "        self.NT_set = set()                            # set of non-terminal symbols\n",
    "        self.T_set = set()                             # set of terminal symbols\n",
    "        self.postags_set = set()                       # set of postags\n",
    "        self.unary_rules_freq = defaultdict(float)     # frequencies of unary rules (A -> B)\n",
    "        self.binary_rules_freq = defaultdict(float)    # frequencies of binary rules (A -> BC)\n",
    "        self.postags_freq = defaultdict(float)         # frequencies of postags (POS -> <word>)\n",
    "        self.unary_rules_dict = {}\n",
    "        self.binary_rules_dict = {}\n",
    "        self.postags_dict = {}\n",
    "        self.not_initialized = True\n",
    "\n",
    "    def initialize(self, NT_set, T_set, postags_set,\n",
    "                   unary_rules_freq, binary_rules_freq, postags_freq,\n",
    "                   unary_rules_dict, binary_rules_dict, postags_dict):\n",
    "        self.NT_set = NT_set\n",
    "        self.T_set = T_set\n",
    "        self.postags_set = postags_set\n",
    "        self.unary_rules_freq = unary_rules_freq\n",
    "        self.binary_rules_freq = binary_rules_freq\n",
    "        self.postags_freq = postags_freq\n",
    "        self.unary_rules_dict = unary_rules_dict\n",
    "        self.binary_rules_dict = binary_rules_dict\n",
    "        self.postags_dict = postags_dict\n",
    "        self.not_initialized = False\n",
    "        \n",
    "    def _parse_sent(self, s, verbose=False):\n",
    "        t0 = time()\n",
    "        \n",
    "        tokens = s.strip().split(u' ')\n",
    "        n = len(tokens)\n",
    "        dp = defaultdict(float)\n",
    "        backPointers = {}\n",
    "        \n",
    "        # POS tagger\n",
    "        for i, w in enumerate(tokens):\n",
    "            if w in self.T_set:\n",
    "                dp[(i, i+1)] = self.postags_dict[w]\n",
    "            else:\n",
    "                dp[(i, i+1)] = self.postags_dict[u\"<UNK>\"]\n",
    "            if verbose > 1:\n",
    "                print (u\"->Add POS tag for {0}:\\n\".format(w))\n",
    "                print (dp[(i, i+1)])\n",
    "            self.add_unary_rules(dp, backPointers, i, i+1, verbose)\n",
    "        \n",
    "        for l in range(2, n + 1):\n",
    "            for i in range(0, n + 1 - l):\n",
    "                j = i + l\n",
    "                dp[(i, j)] = {}\n",
    "                for s in range(i + 1, j):\n",
    "                    B_set = dp[(i, s)]\n",
    "                    C_set = dp[(s, j)]\n",
    "                    for B, prob_B in B_set.iteritems():\n",
    "                        for C, prob_C in C_set.iteritems():\n",
    "                            if (B, C) in self.binary_rules_dict:\n",
    "                                for A, prob_A in self.binary_rules_dict[(B, C)].iteritems():\n",
    "                                    prob = prob_A * prob_B * prob_C\n",
    "                                    if (A not in dp[(i, j)]) or prob > dp[(i, j)][A]:\n",
    "                                        dp[(i, j)][A] = prob\n",
    "                                        backPointers[(i, j, A)] = (s, B, C)\n",
    "                                        if verbose > 1:\n",
    "                                            print (u\"-->Add binary rule ({0}, {1}): {2} -> {3} {4} / {5}\\n\".format(i, j, A, B, C, prob))\n",
    "                self.add_unary_rules(dp, backPointers, i, j, verbose)\n",
    "        \n",
    "        if (0, n, u\"SENT\") not in backPointers:\n",
    "            return None\n",
    "        else:\n",
    "            t = self.buildTree(backPointers, 0, n, u\"SENT\", tokens)\n",
    "            t.un_chomsky_normal_form(expandUnary = False)\n",
    "            return t\n",
    " \n",
    "    def add_unary_rules(self, dp, backPointers, i, j, verbose=False):\n",
    "        B_set = dp[(i, j)].keys()\n",
    "        for B in B_set:\n",
    "            if B in self.unary_rules_dict:\n",
    "                for A, prob_A in self.unary_rules_dict[B].iteritems():\n",
    "                    prob = prob_A * dp[(i, j)][B]\n",
    "                    if (A not in dp[(i, j)]) or prob > dp[(i, j)][A]:\n",
    "                        dp[(i, j)][A] = prob\n",
    "                        backPointers[(i, j, A)] = (B,)\n",
    "                        if verbose > 1:\n",
    "                            print (u\"-->Add unary rule ({0}, {1}): {2} -> {3} / {4}\\n\".format(i, j, A, B, prob))\n",
    "        return\n",
    "    \n",
    "    def buildTree(self, backPointers, i, j, label, tokens):\n",
    "        if (i, j, label) not in backPointers: # Terminals\n",
    "            t = Tree(label, [tokens[i]])\n",
    "        elif len(backPointers[(i, j, label)]) == 1: # Unary rules\n",
    "            child_label = backPointers[(i, j, label)][0]\n",
    "            t = Tree(label, [self.buildTree(backPointers, i, j, child_label, tokens)])\n",
    "        else: # Binary rules\n",
    "            split, child_label0, child_label1 = backPointers[(i, j, label)]\n",
    "            t = Tree(label, [self.buildTree(backPointers, i, split, child_label0, tokens),\n",
    "                             self.buildTree(backPointers, split, j, child_label1, tokens)])\n",
    "        return t\n",
    "    \n",
    "    def parse_sent(self, input, output=None, verbose=False):\n",
    "        if self.not_initialized:\n",
    "            print (\"Parser must be initialized before calling parse function!\")\n",
    "            return\n",
    "        \n",
    "        t0 = time()\n",
    "        \n",
    "        tree = self._parse_sent(input, verbose)\n",
    "        if output == None:\n",
    "            print (tree)\n",
    "        else:\n",
    "            with codecs.open(output, 'w', 'UTF-8') as f:\n",
    "                f.write(u\"{0}\\n\".format(tree))\n",
    "                f.close()\n",
    "                \n",
    "        if verbose:\n",
    "            print (\"Sentence parse done in %0.3fs\" % (time() - t0))\n",
    "        return tree\n",
    "        \n",
    "        \n",
    "    def parse_corpus(self, input, output=None, verbose=False):\n",
    "        if self.not_initialized:\n",
    "            print (\"Parser must be initialized before calling parse function!\")\n",
    "            return\n",
    "        \n",
    "        to = time()\n",
    "        \n",
    "        with codecs.open(input, 'r', 'UTF-8') as f_in:\n",
    "            if output != None:\n",
    "                f_out = codecs.open(output, 'w', 'UTF-8')\n",
    "            data = f_in.read().splitlines()\n",
    "            for sent in data:\n",
    "                tree = self._parse_sent(sent, verbose)\n",
    "                if output != None:\n",
    "                    f_out.write(u\"{0}\\n\".format(tree))\n",
    "                else:\n",
    "                    print (u\"{0}\\n\".format(tree))\n",
    "            f_in.close()\n",
    "            if output != None:\n",
    "                f_out.close()\n",
    "                \n",
    "        if verbose:\n",
    "            print (\"Corpus parse done in %0.3fs\" % (time() - t0))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.14"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}