master-thesis/Tagging/conllu_batch_generator.ipynb

397 lines
13 KiB
Plaintext

{
"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
}