Untitled.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Enter your sentence : hello\n",
      "['PRT']\n",
      "Press 0 if you want to Enter new sentence\n",
      "0\n",
      "Enter your sentence : dgf\n",
      "['.']\n",
      "Press 0 if you want to Enter new sentence\n",
      "\n"
     ]
    }
   ],
   "source": [
    "\n",
    "import numpy as np\n",
    "import random\n",
    "from IPython.core.display import HTML\n",
    "from itertools import chain\n",
    "from collections import Counter, defaultdict\n",
    "from pomegranate import State, HiddenMarkovModel, DiscreteDistribution\n",
    "from collections import namedtuple, OrderedDict\n",
    "\n",
    "Sentence = namedtuple(\"Sentence\", \"words tags\")\n",
    "\n",
    "def read_data(filename):\n",
    "    \"\"\"Read tagged sentence data\"\"\"\n",
    "    with open(filename, 'r') as f:\n",
    "        sentence_lines = [l.split(\"\\n\") for l in f.read().split(\"\\n\\n\")]\n",
    "    return OrderedDict(((s[0], Sentence(*zip(*[l.strip().split(\"\\t\")\n",
    "                        for l in s[1:]]))) for s in sentence_lines if s[0]))\n",
    "\n",
    "\n",
    "def read_tags(filename):\n",
    "    \"\"\"Read a list of word tag classes\"\"\"\n",
    "    with open(filename, 'r') as f:\n",
    "        tags = f.read().split(\"\\n\")\n",
    "    return frozenset(tags)\n",
    "\n",
    "def read_data(filename):\n",
    "    \"\"\"Read tagged sentence data\"\"\"\n",
    "    with open(filename, 'r') as f:\n",
    "        sentence_lines = [l.split(\"\\n\") for l in f.read().split(\"\\n\\n\")]\n",
    "    return OrderedDict(((s[0], Sentence(*zip(*[l.strip().split(\"\\t\")\n",
    "                        for l in s[1:]]))) for s in sentence_lines if s[0]))\n",
    "\n",
    "def read_tags(filename):\n",
    "    \"\"\"Read a list of word tag classes\"\"\"\n",
    "    with open(filename, 'r') as f:\n",
    "        tags = f.read().split(\"\\n\")\n",
    "    return frozenset(tags)\n",
    "\n",
    "class Subset(namedtuple(\"BaseSet\", \"sentences keys vocab X tagset Y N stream\")):\n",
    "    def __new__(cls, sentences, keys):\n",
    "        word_sequences = tuple([sentences[k].words for k in keys])\n",
    "        tag_sequences = tuple([sentences[k].tags for k in keys])\n",
    "        wordset = frozenset(chain(*word_sequences))\n",
    "        tagset = frozenset(chain(*tag_sequences))\n",
    "        N = sum(1 for _ in chain(*(sentences[k].words for k in keys)))\n",
    "        stream = tuple(zip(chain(*word_sequences), chain(*tag_sequences)))\n",
    "        return super().__new__(cls, {k: sentences[k] for k in keys}, keys, wordset, word_sequences,\n",
    "                               tagset, tag_sequences, N, stream.__iter__)\n",
    "\n",
    "    def __len__(self):\n",
    "        return len(self.sentences)\n",
    "\n",
    "    def __iter__(self):\n",
    "        return iter(self.sentences.items())\n",
    "\n",
    "\n",
    "class Dataset(namedtuple(\"_Dataset\", \"sentences keys vocab X tagset Y training_set testing_set N stream\")):\n",
    "    def __new__(cls, tagfile, datafile, train_test_split=0.8, seed=112890):  \n",
    "        tagset = read_tags(tagfile)\n",
    "        sentences = read_data(datafile)\n",
    "        keys = tuple(sentences.keys())\n",
    "        wordset = frozenset(chain(*[s.words for s in sentences.values()]))\n",
    "        word_sequences = tuple([sentences[k].words for k in keys])\n",
    "        tag_sequences = tuple([sentences[k].tags for k in keys])\n",
    "        N = sum(1 for _ in chain(*(s.words for s in sentences.values())))\n",
    "        \n",
    "        # split data into train/test sets\n",
    "        _keys = list(keys)\n",
    "        if seed is not None: random.seed(seed)\n",
    "        random.shuffle(_keys)\n",
    "        split = int(train_test_split * len(_keys))\n",
    "        training_data = Subset(sentences, _keys[:split])\n",
    "        testing_data = Subset(sentences, _keys[split:])\n",
    "        stream = tuple(zip(chain(*word_sequences), chain(*tag_sequences)))\n",
    "        return super().__new__(cls, dict(sentences), keys, wordset, word_sequences, tagset,\n",
    "                               tag_sequences, training_data, testing_data, N, stream.__iter__)\n",
    "\n",
    "    def __len__(self):\n",
    "        return len(self.sentences)\n",
    "\n",
    "    def __iter__(self):\n",
    "        return iter(self.sentences.items())\n",
    "\n",
    "data = Dataset(r\"D:\\Programming\\Data_Science_Project\\tags-universal.txt\", r\"D:\\Programming\\Data_Science_Project\\brown-universal.txt\", train_test_split=0.7)\n",
    "\n",
    "assert len(data) == len(data.training_set) + len(data.testing_set), \\\n",
    "       \"The number of sentences in the training set + testing set should sum to the number of sentences in the corpus\"\n",
    "\n",
    "def pair_counts(tags, words):\n",
    "    d = defaultdict(lambda: defaultdict(int))\n",
    "    for tag, word in zip(tags, words):\n",
    "        d[tag][word] += 1\n",
    "        \n",
    "    return d\n",
    "\n",
    "def unigram_counts(sequences):\n",
    "\n",
    "    return Counter(sequences)\n",
    "\n",
    "def bigram_counts(sequences):\n",
    "\n",
    "    d = Counter(sequences)\n",
    "    return d\n",
    "\n",
    "tags = [tag for i, (word, tag) in enumerate(data.stream())]         #Collects the tags\n",
    "o = [(tags[i],tags[i+1]) for i in range(0,len(tags)-2,2)]           #collects pairs of tags\n",
    "tag_bigrams = bigram_counts(o) \n",
    "\n",
    "def starting_counts(sequences):\n",
    "    \n",
    "    d = Counter(sequences)\n",
    "    return d\n",
    "\n",
    "def ending_counts(sequences):\n",
    "    \n",
    "    d = Counter(sequences)\n",
    "    return d\n",
    "\n",
    "#Model Accuracy Evaluation\n",
    "\n",
    "def accuracy(X, Y, model):\n",
    "    \n",
    "    correct = total_predictions = 0\n",
    "    for observations, actual_tags in zip(X, Y):\n",
    "        \n",
    "        # The model.viterbi call in simplify_decoding will return None if the HMM\n",
    "        # raises an error (for example, if a test sentence contains a word that\n",
    "        # is out of vocabulary for the training set). Any exception counts the\n",
    "        # full sentence as an error (which makes this a conservative estimate).\n",
    "        try:\n",
    "            most_likely_tags = simplify_decoding(observations, model)\n",
    "            correct += sum(p == t for p, t in zip(most_likely_tags, actual_tags))\n",
    "        except:\n",
    "            pass\n",
    "        total_predictions += len(observations)\n",
    "    return correct / total_predictions\n",
    "\n",
    "\n",
    "#IMPLEMENTATION: Basic HMM Tagger\n",
    "basic_model = HiddenMarkovModel(name=\"base-hmm-tagger\")\n",
    "\n",
    "tags = [tag for i, (word, tag) in enumerate(data.stream())]            # tags in whole corpus\n",
    "words = [word for i, (word, tag) in enumerate(data.stream())]          # words in whole corpus\n",
    "\n",
    "tags_count=unigram_counts(tags)                                        #Counts the no. of tags in whole corpus \"tag_unigrams\"\n",
    "tag_words_count=pair_counts(tags,words)                     #Give count of a particular tags with word appeared in sentence\n",
    "\n",
    "starting_tag_list=[i[0] for i in data.Y]\n",
    "ending_tag_list=[i[-1] for i in data.Y]\n",
    "\n",
    "starting_tag_count=starting_counts(starting_tag_list)            #the number of times a tag occured at the start\n",
    "ending_tag_count=ending_counts(ending_tag_list)                  #the number of times a tag occured at the end\n",
    "\n",
    "\n",
    "\n",
    "to_pass_states = []\n",
    "for tag, words_dict in tag_words_count.items():\n",
    "    total = float(sum(words_dict.values()))\n",
    "    distribution = {word: count/total for word, count in words_dict.items()}\n",
    "    tag_emissions = DiscreteDistribution(distribution)\n",
    "    tag_state = State(tag_emissions, name=tag)\n",
    "    to_pass_states.append(tag_state)\n",
    "\n",
    "\n",
    "basic_model.add_states()    \n",
    "    \n",
    "\n",
    "start_prob={}\n",
    "\n",
    "for tag in tags:\n",
    "    start_prob[tag]=starting_tag_count[tag]/tags_count[tag]\n",
    "\n",
    "for tag_state in to_pass_states :\n",
    "    basic_model.add_transition(basic_model.start,tag_state,start_prob[tag_state.name])    \n",
    "\n",
    "end_prob={}\n",
    "\n",
    "for tag in tags:\n",
    "    end_prob[tag]=ending_tag_count[tag]/tags_count[tag]\n",
    "for tag_state in to_pass_states :\n",
    "    basic_model.add_transition(tag_state,basic_model.end,end_prob[tag_state.name])\n",
    "    \n",
    "\n",
    "\n",
    "transition_prob_pair={}\n",
    "\n",
    "for key in tag_bigrams.keys():\n",
    "    transition_prob_pair[key]=tag_bigrams.get(key)/tags_count[key[0]]\n",
    "for tag_state in to_pass_states :\n",
    "    for next_tag_state in to_pass_states :\n",
    "        basic_model.add_transition(tag_state,next_tag_state,transition_prob_pair[(tag_state.name,next_tag_state.name)])\n",
    "\n",
    "basic_model.bake()\n",
    "\n",
    "#Prediction Making\n",
    "\n",
    "def replace_unknown(sequence):\n",
    "    \n",
    "    return [w if w in data.training_set.vocab else 'nan' for w in sequence]\n",
    "\n",
    "def simplify_decoding(X, model):\n",
    "    \n",
    "    _, state_path = model.viterbi(replace_unknown(X))\n",
    "    return [state[1].name for state in state_path[1:-1]]\n",
    "\n",
    "#Visual inspection of model accuracy\n",
    "\n",
    "#for key in data.testing_set.keys[2:3]:\n",
    "#    print(\"Sentence Key: {}\\n\".format(key))\n",
    "#    print(\"Predicted labels:\\n-----------------\")\n",
    "#    print(simplify_decoding(data.sentences[key].words, basic_model))\n",
    "#    print()\n",
    "#    print(\"Actual labels:\\n--------------\")\n",
    "#    print(data.sentences[key].tags)\n",
    "#    print(\"\\n\")\n",
    "\n",
    "#Overall accuracy of our model\n",
    "#hmm_training_acc = accuracy(data.training_set.X, data.training_set.Y, basic_model)\n",
    "#print(\"training accuracy basic hmm model: {:.2f}%\".format(100 * hmm_training_acc))\n",
    "\n",
    "#hmm_testing_acc = accuracy(data.testing_set.X, data.testing_set.Y, basic_model)\n",
    "#print(\"testing accuracy basic hmm model: {:.2f}%\".format(100 * hmm_testing_acc))\n",
    "x=0\n",
    "\n",
    "while(x == 0):\n",
    "    \n",
    "    sentence = input(\"Enter your sentence : \")\n",
    "    formate=list(sentence.split())\n",
    "    result1 =simplify_decoding(formate, basic_model)\n",
    "    print(result1)\n",
    "    file = open(\"result.txt\",\"a+\")\n",
    "    \n",
    "    for i,j in zip(formate,result1):\n",
    "        file.write(i)\n",
    "        file.write(\" : \")\n",
    "        file.write(j)\n",
    "        file.write('\\n')\n",
    "    \n",
    "    \n",
    "    try:\n",
    "        x = int(input(\"Press 0 if you want to Enter new sentence\\n\"))\n",
    "    except ValueError:\n",
    "        x = 1\n",
    "    if x == 0:\n",
    "        file.write('\\nNew Sentence :\\n\\n')\n",
    "    else:\n",
    "        file.close()\n",
    "        "
   ]
  },
  {
   "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.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}