{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Conllu Batch Generator\n", "\n", "read conllu documents in batches" ] }, { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [], "source": [ "import sys\n", "sys.path.append('../')\n", "\n", "from conllu import parse\n", "from Tagging.tagging_tools import print_visualized_tags\n", "\n", "from sklearn import preprocessing\n", "import numpy as np\n", "\n", "\n", "import settings # noqa\n", "\n", "import gzip" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "class ConlluSentenceIterator(object):\n", " def __init__(self, conllu_reader):\n", " self.conllu_reader = conllu_reader\n", " self._fileobj = None\n", " self._open()\n", " \n", " def _open(self):\n", " if self.conllu_reader._path.endswith(\".gz\"):\n", " self._fileobj = gzip.open(self.conllu_reader._path, 'r')\n", " self._nextline = self.read_byte_line\n", " else:\n", " self._fileobj = open(self.conllu_reader._path, 'r')\n", " self._nextline = self.read_str_line\n", "\n", " def __next__(self):\n", " next_sent = self.next_sentence()\n", " if next_sent is None:\n", " raise StopIteration\n", " return next_sent\n", " \n", " def read_str_line(self):\n", " return self._fileobj.readline()\n", " \n", " def read_byte_line(self):\n", " return self._fileobj.readline().decode(\"utf-8\")\n", "\n", " def next_sentence(self):\n", " data = \"\"\n", " while True:\n", " line = self._nextline()\n", " if line == \"\":\n", " break\n", " if line == \"\\n\" and len(data) > 0:\n", " break\n", " data += line\n", "\n", " if data == \"\":\n", " return None\n", "\n", " if data[-1] != \"\\n\":\n", " data += \"\\n\"\n", "\n", " conllu_obj = parse(data + \"\\n\")\n", " return conllu_obj" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "class ConlluDocumentIterator(object):\n", " def __init__(self, conllu_reader, return_recipe_ids = False):\n", " self.conllu_reader = conllu_reader\n", " self._fileobj = None\n", " self._open()\n", " self._return_recipe_ids = return_recipe_ids\n", " \n", " def _open(self):\n", " if self.conllu_reader._path.endswith(\".gz\"):\n", " self._fileobj = gzip.open(self.conllu_reader._path, 'r')\n", " self._nextline = self.read_byte_line\n", " else:\n", " self._fileobj = open(self.conllu_reader._path, 'r')\n", " self._nextline = self.read_str_line\n", " \n", " def read_str_line(self):\n", " return self._fileobj.readline()\n", " \n", " def read_byte_line(self):\n", " return self._fileobj.readline().decode(\"utf-8\")\n", "\n", " def next_document(self):\n", " doc_id = None\n", " data = \"\"\n", " last_line_empty = False\n", " while True:\n", " line = self._nextline()\n", " if line.startswith('#'):\n", " # looking for an recipe id:\n", " comment = line.replace('#', '')\n", " splitted = comment.split(':')\n", " if len(splitted) == 2:\n", " if splitted[0].strip() == \"id\":\n", " doc_id = splitted[1].strip()\n", " continue\n", " \n", " if line == \"\":\n", " break\n", " if line == \"\\n\" and len(data) > 0:\n", " if last_line_empty:\n", " break\n", " last_line_empty = True\n", " else:\n", " last_line_empty = False\n", " data += line\n", "\n", " if data == \"\":\n", " return None\n", "\n", " if data[-1] != \"\\n\":\n", " data += \"\\n\"\n", "\n", " conllu_obj = parse(data + \"\\n\")\n", " \n", " if self._return_recipe_ids:\n", " return conllu_obj, doc_id\n", " return conllu_obj\n", "\n", " def __next__(self):\n", " next_sent = self.next_document()\n", " if next_sent is None:\n", " raise StopIteration\n", " return next_sent" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "class ConlluReader(object):\n", " def __init__(self, path, iter_documents=False, return_recipe_ids = False):\n", " self._path = path\n", " self.iter_documents = iter_documents\n", " self.return_recipe_ids = return_recipe_ids\n", "\n", " def __iter__(self):\n", " return ConlluDocumentIterator(self, self.return_recipe_ids) if self.iter_documents else ConlluSentenceIterator(self)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "class SlidingWindowListIterator(object):\n", " def __init__(self, parent):\n", " self.parent = parent\n", " self.i = 0\n", "\n", " def __next__(self):\n", " if len(self.parent) == self.i:\n", " raise StopIteration\n", "\n", " self.i += 1\n", " return self.parent[self.i - 1]" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "class SlidingWindowList(list):\n", " def __init__(self, sliding_window_size, input=None, border_value=None):\n", "\n", " self.sliding_window_size = sliding_window_size\n", " self.border_value = border_value\n", "\n", " if border_value is None and input is not None:\n", " self.border_value = type(input[0])()\n", "\n", " if input is not None:\n", " super(SlidingWindowList, self).__init__(input)\n", "\n", " def __getitem__(self, index):\n", "\n", " if type(index) == slice:\n", " start = 0 if index.start is None else index.start\n", " stop = len(self) if index.stop is None else index.stop\n", " step = 1 if index.step is None else index.step\n", " return [self[i] for i in range(start, stop, step)]\n", "\n", " else:\n", " n = self.sliding_window_size * 2 + 1\n", " res = n * [self.border_value]\n", "\n", " j_start = index - self.sliding_window_size\n", "\n", " for i in range(n):\n", " ind = j_start + i\n", " if ind >= 0 and ind < len(self):\n", " res[i] = super(SlidingWindowList, self).__getitem__(ind)\n", "\n", " return res\n", "\n", " def __iter__(self):\n", " return SlidingWindowListIterator(self)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "'''\n", "class ConlluDataProviderIterator(object):\n", " def __init__(self, parent):\n", " self.parent = parent\n", " self.conllu_reader = ConlluReader(\n", " parent.filepath, parent.iter_documents)\n", "\n", " def __next__(self):\n", " result = self.parent.getNextDataBatch(conllu_reader=self.conllu_reader)\n", " if result is None:\n", " raise StopIteration\n", " return result\n", "'''\n", "\n", "'''\n", "class ConlluDataProvider(object):\n", " def __init__(self,\n", " filepath,\n", " word2vec_model,\n", " batchsize=100,\n", " window_size=3,\n", " iter_documents=False,\n", " food_type=None):\n", " self.batchsize = batchsize\n", " self.word2vec_model = word2vec_model\n", " self.filepath = filepath\n", " self.conllu_reader = ConlluReader(filepath, iter_documents)\n", " self.window_size = window_size\n", " self.food_type = food_type\n", " self.iter_documents = iter_documents\n", "\n", " # create a label binarizer for upos tags:\n", " self.lb = preprocessing.LabelBinarizer()\n", " self.lb.fit(['.', 'ADJ', 'ADP', 'ADV', 'CONJ', 'DET',\n", " 'NOUN', 'NUM', 'PRON', 'PRT', 'VERB', 'X'])\n", "\n", " def _get_next_conllu_objects(self, n: int, conllu_reader):\n", " i = 0\n", " conllu_list = []\n", "\n", " while i < n:\n", " try:\n", " conllu_list.append(conllu_reader.__iter__().__next__())\n", " i += 1\n", "\n", " except StopIteration:\n", " break\n", "\n", " return conllu_list\n", "\n", " def _get_upos_X(self, conllu_list):\n", " n_tokens = 0\n", " l_global = []\n", " for document in conllu_list:\n", " l = []\n", " for sentence in document:\n", " for token in sentence:\n", " upos = token['upostag']\n", " l.append(upos)\n", " n_tokens += 1\n", " if len(l) > 0:\n", " l_global.append(self.lb.transform(l))\n", "\n", " return l_global, n_tokens\n", "\n", " def _get_y(self, conllu_list, misk_key=\"food_type\", misc_val=\"ingredient\"):\n", " n_tokens = 0\n", " y_global = []\n", " for document in conllu_list:\n", " y = []\n", " for sentence in document:\n", " for token in sentence:\n", " m = token['misc']\n", " t_y = m is not None and misk_key in m and m[misk_key] == misc_val\n", " y.append(t_y)\n", " n_tokens += 1\n", " if len(y) > 0:\n", " y_global.append(y)\n", "\n", " return y_global, n_tokens\n", "\n", " def getNextDataBatch(self, y_food_type_label=None, conllu_reader=None):\n", "\n", " if y_food_type_label is None:\n", " y_food_type_label = self.food_type\n", "\n", " if conllu_reader is None:\n", " conllu_reader = self.conllu_reader\n", " conllu_list = self._get_next_conllu_objects(\n", " self.batchsize, conllu_reader)\n", "\n", " if len(conllu_list) == 0:\n", " return None\n", "\n", " # generate features for each document/sentence\n", " n = len(conllu_list)\n", "\n", " d = self.window_size * 2 + 1\n", "\n", " buf_X, x_tokens = self._get_upos_X(conllu_list)\n", " buf_ingr_y, y_tokens = self._get_y(conllu_list)\n", "\n", " assert len(buf_X) == len(buf_ingr_y) and x_tokens == y_tokens\n", "\n", " X_upos = np.zeros(shape=(x_tokens, d * len(self.lb.classes_)))\n", " y = None\n", "\n", " if y_food_type_label is not None:\n", " y = np.zeros(shape=(x_tokens))\n", "\n", " i = 0\n", " for xupos in buf_X:\n", " tmp = SlidingWindowList(self.window_size,\n", " xupos,\n", " border_value=[0] * len(self.lb.classes_))\n", " for upos_window in tmp:\n", " X_upos[i, :] = np.array(upos_window).flatten()\n", " i += 1\n", "\n", " i = 0\n", " if y_food_type_label is not None:\n", " for sentence in buf_ingr_y:\n", " for yl in sentence:\n", " y[i] = yl\n", " i += 1\n", "\n", " return X_upos, y\n", " \n", " def __iter__(self):\n", " return ConlluDataProviderIterator(self)\n", "\n", "'''" ] } ], "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.3" } }, "nbformat": 4, "nbformat_minor": 4 }