-
-
Notifications
You must be signed in to change notification settings - Fork 31
/
dataset_container.py
311 lines (248 loc) · 13.8 KB
/
dataset_container.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
# pylint: disable=too-many-arguments
from abc import ABC, abstractmethod
from pickle import load
from typing import Union, List, Dict, Callable, Tuple
import pandas as pd
from torch.utils.data import Dataset
from .tools import former_python_list, validate_column_names
from ..errors.data_error import DataError
from ..data_validation import validate_if_any_empty, validate_if_any_whitespace_only, validate_if_any_none
class DatasetContainer(Dataset, ABC):
"""
Interface for the dataset. This interface defines most of the methods that the dataset needs to define.
If you define another dataset container, the init must define the attribute data.
We also recommend using the ``validate_dataset`` method in your init to validate some characteristics of your
dataset.
For a training container, it validates the following:
- all addresses are not None value,
- all addresses are not empty,
- all addresses are not whitespace string,
- all tags are not empty, if data is a list of tuple (``[('an address', ['a_tag', 'another_tag']), ...]``), and
- if the addresses (whitespace-split) are the same length as their respective tags list.
While for a predict container (unknown prediction tag), it validates the following:
- all addresses are not None,
- all addresses are not empty, and
- all addresses are not whitespace string.
Args:
is_training_container (bool): Either or not, the dataset container is a training container. This will determine
the dataset validation test we apply to the dataset. That is, a predict dataset doesn't include tags.
The default value is true.
"""
@abstractmethod
def __init__(self, is_training_container: bool = True) -> None:
"""
Need to be defined by the child class.
"""
self.data = None
self.is_training_container = is_training_container
def __len__(self) -> int:
return len(self.data)
def __getitem__(
self, idx: Union[int, slice]
) -> Union[List[str], str, List[List[Tuple[str, List]]], Tuple[str, List]]:
"""
If the DatasetContainer is a "predict" one:
- it can be a list of string items (e.g. a list of addresses (str)), or
- it can be a unique string item (e.g. one address).
If the DatasetContainer is a training one:
- it can be a list of tuple (str, list) items, namely a list of parsed examples (e.g. an address with
the tags), or
- it can be a tuple (str, list) item.
"""
if isinstance(idx, slice):
start, stop, _ = idx.indices(len(self))
result = [self.data[index] for index in range(start, stop)]
else:
result = self.data[idx]
return result
def is_a_train_container(self) -> bool:
return self.is_training_container
def validate_dataset(self) -> None:
if not self._data_is_a_list():
raise TypeError("The dataset is not a list.")
if self.is_training_container:
data_to_validate = [data[0] for data in self.data]
else:
data_to_validate = self.data
if validate_if_any_none(string_elements=data_to_validate):
raise DataError("Some addresses data points are None value.")
if self.is_training_container:
# Not done in previous similar if since none test not applied
self._training_validation()
if validate_if_any_empty(string_elements=data_to_validate):
raise DataError("Some addresses data points are empty.")
if validate_if_any_whitespace_only(string_elements=data_to_validate):
raise DataError("Some addresses only include whitespace thus cannot be parsed.")
def _data_is_a_list(self) -> bool:
return isinstance(self.data, list)
def _training_validation(self) -> None:
if not self._data_is_list_of_tuple():
raise TypeError(
"The pickled dataset data are not in a tuple format. Data"
"is expected to be a list of tuples where the first element is"
"the address, and the second is the address tag."
)
if self._empty_tags():
raise DataError("Some tags data points are empty.")
if not self._data_tags_is_same_len_then_address():
print(
f"Some addresses (whitespace-split) and the associated tags are not the same len. "
f"If you are using a CSVDatasetContainer, consider using the tag_seperator_reformat_fn argument."
f"Here is the report of those cases where len differ to help you out:\n"
f"{self._data_tags_not_the_same_len_diff()}"
)
raise DataError("Some addresses (whitespace-split) and the tags associated with them are not the same len.")
def _data_is_list_of_tuple(self) -> bool:
"""
Return true if one of the elements in the dataset is not a tuple.
"""
return all(isinstance(data, tuple) for data in self.data)
def _empty_tags(self) -> bool:
"""
Return true if one of the tag sets is empty.
"""
return all(len(data[1]) == 0 for data in self.data)
def _data_tags_is_same_len_then_address(self) -> bool:
"""
Return true if all the data tags are the same len as the address split at each whitespace.
"""
return all(len(data[0].split(" ")) == len(data[1]) for data in self.data)
def _data_tags_not_the_same_len_diff(self) -> str:
diff_indexes = [index for index, data in enumerate(self.data) if len(data[0].split(" ")) != len(data[1])]
report = ""
for diff_index in diff_indexes:
report += (
f"The data point (with index: {diff_index}: {self.data[diff_index]}"
f"\n\tLen of the address: {len(self.data[diff_index][0].split(' '))}"
f"\n\tLen of the tags: {len(self.data[diff_index][1])}\n"
)
return report
class PickleDatasetContainer(DatasetContainer):
"""
Pickle dataset container that imports a list of addresses in pickle format and does some validation on it.
The dataset needs to be a list of tuples where the first element of each tuple is the address (a string),
and the second is a list of the expected tag to predict (e.g. ``[('an address', ['a_tag', 'another_tag']), ...]``).
The len of the tags needs to be the same as the len of the address when whitespace split.
For a training container, the validation tests applied on the dataset are the following:
- all addresses are not None value,
- all addresses are not empty,
- all addresses are not whitespace string,
- all tags are not empty, if data is a list of tuple (``[('an address', ['a_tag', 'another_tag']), ...]``), and
- if the addresses (whitespace-split) are the same length as their respective tags list.
While for a predict container (unknown prediction tag), the validation tests applied on the dataset are the
following:
- all addresses are not None value,
- all addresses are not empty, and
- all addresses are not whitespace string.
Args:
data_path (str): The path to the pickle dataset file.
is_training_container (bool): Either or not, the dataset container is a training container. This will determine
the dataset validation test we apply to the dataset. That is, a predict dataset doesn't include tags.
The default value is true.
"""
def __init__(self, data_path: str, is_training_container: bool = True) -> None:
super().__init__(is_training_container=is_training_container)
with open(data_path, "rb") as f:
self.data = load(f)
if not is_training_container:
if self._test_predict_container_is_list_of_tuple():
raise DataError(
"The data is a list of tuples, but the dataset container is a predict container. "
"Predict container should contain only a list of addresses."
)
self.validate_dataset()
def _test_predict_container_is_list_of_tuple(self) -> bool:
return any((isinstance(data, tuple) for data in self.data))
class CSVDatasetContainer(DatasetContainer):
"""
CSV dataset container that imports a CSV of addresses. If the dataset is a predict one, it needs to have at least
one column with some addresses. If the dataset is a training one (with prediction tags), it needs to have at
least two columns, one with some addresses and another with a list of tags for each address.
After loading the CSV dataset, some tests will be applied depending on its type.
For a training container, the validation tests applied on the dataset are the following:
- all addresses are not None value,
- all addresses are not empty,
- all addresses are not whitespace string, and
- if the addresses (whitespace-split) are the same length as their respective tags list.
While for a predict container (unknown prediction tag), the validation tests applied on the dataset are the
following:
- all addresses are not None value,
- all addresses are not empty, and
- all addresses are not whitespace string.
Args:
data_path (str): The path to the CSV dataset file.
column_names (list): A column name list to extract the dataset element.
If the dataset container is a "predict" one, the list must be of exactly one element
(i.e. the address column). On the other hand, if the dataset container is a "training" one, the list must be
of exactly two elements: addresses and tags.
is_training_container (bool): Either or not, the dataset container is a training container. This will determine
the dataset validation test we apply to the dataset. That is, a predict dataset doesn't include tags.
The default value is true.
separator (str): The CSV columns separator to use. By default, ``"\\t"``.
tag_seperator_reformat_fn (Callable, optional): A function to parse a tags string and return a list of
address tags. For example, if the tag column is a former Python list saved with pandas, the characters ``]``
, ``]`` and ``'`` will be included as the tags' element. Thus, a parsing function will take a string as is
parameter and output a python list. The default function processes it as a former Python list.
That is, it removes the ``[],`` characters and splits the sequence at each comma (``","``).
csv_reader_kwargs (dict, optional): Keyword arguments to pass to pandas ``read_csv`` use internally. By default,
the ``data_path`` is passed along with our default ``sep`` value ( ``"\\t"``) and the ``"utf-8"`` encoding
format. However, this can be overridden by using this argument again.
"""
def __init__(
self,
data_path: str,
column_names: Union[List, str],
is_training_container: bool = True,
separator: str = "\t",
tag_seperator_reformat_fn: Union[None, Callable] = None,
csv_reader_kwargs: Union[None, Dict] = None,
) -> None:
super().__init__(is_training_container=is_training_container)
if is_training_container:
if isinstance(column_names, str):
raise ValueError(
"When the dataset is a training container, the column names should be a list of column name."
)
if len(column_names) != 2:
raise ValueError("When the dataset is a training container, two column names must be provided.")
else: # It means it is a predict container
if isinstance(column_names, str):
# We transform the str into a list to assess is len
column_names = [column_names]
if len(column_names) != 1:
raise ValueError("When the dataset is a predict container, one column name must be provided.")
if validate_column_names(column_names):
raise ValueError("A column name is an empty string or whitespace only.")
if csv_reader_kwargs is None:
csv_reader_kwargs = {}
csv_reader_kwargs = {
"filepath_or_buffer": data_path,
"sep": separator,
"encoding": "utf-8",
**csv_reader_kwargs,
}
if tag_seperator_reformat_fn is None:
tag_seperator_reformat_fn = former_python_list
if is_training_container:
data = [
(data_point[0], tag_seperator_reformat_fn(data_point[1]))
for data_point in pd.read_csv(**csv_reader_kwargs)[column_names].to_numpy()
]
else:
data = [data_point[0] for data_point in pd.read_csv(**csv_reader_kwargs)[column_names].to_numpy()]
self.data = data
self.validate_dataset()
class ListDatasetContainer(DatasetContainer):
"""
List dataset container that loads a list dataset into a DatasetContainer class. It also validates the dataset.
Args:
data (list): The dataset in a list format. The list format (if a train or test container) is
identical as the :class:`~deepparse.dataset_container.PickleDatasetContainer`.
is_training_container (bool): Either or not, the dataset container is a training container. This will determine
the dataset validation test we apply to the dataset. That is, a predict dataset doesn't include tags.
The default value is true.
"""
def __init__(self, data: List, is_training_container: bool = True) -> None:
super().__init__(is_training_container=is_training_container)
self.data = data
self.validate_dataset()