pos_crf2_input.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "pos_crf2_input.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "machine_shape": "hm"
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    }
  },
  "cells": [
    {
      "cell_type": "code",
      "metadata": {
        "id": "9mkKmFgHbAeu",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "!pip install sklearn_crfsuite"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "r8yrXJk_YPF-",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import re\n",
        "import random\n",
        "\n",
        "\n",
        "class DataLoader:\n",
        "    def __init__(self):         \n",
        "        sents = self.data_reader('/content/drive/My Drive/Colab Notebooks/ezafe/data/bijankhan_corpus.tsv')\n",
        "        \n",
        "        random.seed(17)\n",
        "        random.shuffle(sents)\n",
        "        data_split_1 = int(len(sents) * .1)\n",
        "        data_split_2 = int(len(sents) * .2)\n",
        "\n",
        "        self.test_data = sents[:data_split_1]\n",
        "        self.dev_data = sents[data_split_1:data_split_2]\n",
        "        self.train_data = sents[data_split_2:]\n",
        "\n",
        "\n",
        "    def data_reader(self, directory):\n",
        "        sents, sent = [], []\n",
        "        with open(directory) as corpus:\n",
        "            for line in corpus:\n",
        "                if line != '\\n':\n",
        "                    word, tag, ezafe_tag = line.strip().split()\n",
        "                    word = word.replace('ي', 'ی').replace('ك', 'ک').replace('ة', 'ه')\n",
        "                    sent.append(((word, int(ezafe_tag)), tag))\n",
        "                else:\n",
        "                    sents.append(sent)\n",
        "                    sent = []\n",
        "        \n",
        "        return sents\n",
        "\n",
        "    def data_loader(self):\n",
        "        return self.train_data, self.dev_data, self.test_data"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "6BT86qVDY7gK",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import pickle\n",
        "\n",
        "from nltk.tag.util import untag\n",
        "from sklearn_crfsuite import CRF\n",
        "from sklearn_crfsuite import metrics\n",
        "\n",
        "\n",
        "def features(sentence, index):\n",
        "    ezafe_tags = [x[1] for x in sentence]\n",
        "    sentence = [x[0] for x in sentence]\n",
        "    \"\"\" sentence: [w1, w2, ...], index: the index of the word \"\"\"\n",
        "    return {\n",
        "        'word': sentence[index],\n",
        "        'has_ezafe': bool(ezafe_tags[index]),\n",
        "        'is_first': index == 0,\n",
        "        'is_last': index == len(sentence) - 1,\n",
        "        'is_2_to_last': index == len(sentence) - 2,\n",
        "        'prefix-1': sentence[index][0],\n",
        "        'prefix-2': sentence[index][:2],\n",
        "        'prefix-3': sentence[index][:3],\n",
        "        'suffix-1': sentence[index][-1],\n",
        "        'suffix-2': sentence[index][-2:],\n",
        "        'suffix-3': sentence[index][-3:],\n",
        "        'prev_word': '' if index == 0 else sentence[index - 1],\n",
        "        '2_prev_word': '' if index == 0 or index == 1 else sentence[index - 2],\n",
        "        '3_prev_word': '' if index == 0 or index == 1 or index == 2 else sentence[index - 3],\n",
        "        '4_prev_word': '' if index in range(0, 4) else sentence[index - 4],\n",
        "        '5_prev_word': '' if index in range(0, 5) else sentence[index - 5],\n",
        "        'next_word': '' if index >= (len(sentence) - 1) else sentence[index + 1],\n",
        "        'next_word_2': '' if index >= (len(sentence) - 2) else sentence[index + 2],\n",
        "        'next_word_3': '' if index >= (len(sentence) - 3) else sentence[index + 3],\n",
        "        'next_word_4': '' if index >= (len(sentence) - 4) else sentence[index + 4],\n",
        "        'next_word_5': '' if index >= (len(sentence) - 5) else sentence[index + 5],\n",
        "    }\n",
        "\n",
        "# transform the dataset from [[('ali', 'N'), ('be', 'P'), ('madrese', 'N'), ('raft', 'V')]]\n",
        "# to [features], [tags]\n",
        "def transform_to_dataset(tagged_sentences):\n",
        "\tX, y = [], []\n",
        "\tfor tagged in tagged_sentences:\n",
        "\t\tX.append([features(untag(tagged), index) for index in range(len(tagged))])\n",
        "\t\ty.append([tag for _, tag in tagged])\n",
        "\t\n",
        "\treturn X, y\n",
        "\n",
        "# instantiating the DataLoader module and loading the data\n",
        "data_loader = DataLoader()\n",
        "training_sentences, dev_sentences, test_sentences = data_loader.data_loader()\n",
        "\n",
        "print(training_sentences[0])\n",
        "\n",
        "X_train, y_train = transform_to_dataset(training_sentences)\n",
        "X_dev, y_dev = transform_to_dataset(dev_sentences)\n",
        "X_test, y_test = transform_to_dataset(test_sentences)\n",
        "\n",
        "# some printing for the sake of debugging\n",
        "print(len(X_train))     \n",
        "print(len(X_test))\n",
        "print(X_train[0])\n",
        "print(y_train[0])\n",
        "\n",
        "# defining the model\n",
        "model = CRF(algorithm='lbfgs',\n",
        "    \t      c1=0.1,\n",
        "            c2=0.1,\n",
        "            max_iterations=100,\n",
        "            all_possible_transitions=True)\n",
        "\n",
        "# training the model\n",
        "model.fit(X_train, y_train)\n",
        "  \n",
        "# validating the model\n",
        "y_pred = model.predict(X_dev)\n",
        "print(metrics.flat_classification_report(y_dev, y_pred, digits=4))\n",
        "\n",
        "# saving the model\n",
        "with open('crf_model.pickle', 'wb') as handle:\n",
        "\t pickle.dump(model, handle, protocol=pickle.HIGHEST_PROTOCOL)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "eCbAPel9NNj7",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# testing the model\n",
        "y_pred = model.predict(X_test)\n",
        "print(metrics.flat_classification_report(y_test, y_pred, digits=4))"
      ],
      "execution_count": 0,
      "outputs": []
    }
  ]
}