"""This file and its contents are licensed under the Apache License 2.0. Please see the included NOTICE for copyright information and LICENSE for a copy of the license.
|
"""
|
import logging
|
from functools import reduce
|
from operator import getitem
|
from urllib.parse import urlparse
|
|
import ujson as json
|
from core.label_config import replace_task_data_undefined_with_config_field
|
from rest_framework.exceptions import ValidationError
|
|
|
class SkipField(Exception):
|
pass
|
|
|
_DATA_TYPES = {
|
'Text': [str, int, float, list],
|
'Header': [str, int, float],
|
'HyperText': [str],
|
'Image': [str, list],
|
'Paragraphs': [list, str],
|
'Chat': [list, str],
|
'Table': [dict, list, str],
|
'TimeSeries': [dict, list, str],
|
'TimeSeriesChannel': [dict, list, str],
|
'List': [list, str],
|
'Choices': [str, list],
|
'PolygonLabels': [str, list],
|
'Labels': [str, list],
|
'BrushLabels': [str, list],
|
'EllipseLabels': [str, list],
|
'HyperTextLabels': [str, list],
|
'KeyPointLabels': [str, list],
|
'ParagraphLabels': [str, list],
|
'RectangleLabels': [str, list],
|
'TimeSeriesLabels': [str, list],
|
'Taxonomy': [str, list, type(None)],
|
'Ranker': [list, str],
|
}
|
logger = logging.getLogger(__name__)
|
|
|
class TaskValidator:
|
"""Task Validator with project scheme configs validation. It is equal to TaskSerializer from django backend."""
|
|
def __init__(self, project, instance=None):
|
self.project = project
|
self.instance = instance
|
self.annotation_count = 0
|
self.prediction_count = 0
|
|
@staticmethod
|
def check_data(project, data):
|
"""Validate data from task['data']"""
|
if data is None:
|
raise ValidationError('Task is empty (None)')
|
|
replace_task_data_undefined_with_config_field(data, project)
|
|
# iterate over data types from project
|
for data_key, data_type in project.data_types.items():
|
|
# get array name in case of Repeater tag
|
is_array = '[' in data_key
|
data_key = data_key.split('[')[0]
|
|
if '.' in data_key:
|
keys = data_key.split('.')
|
try:
|
data_item = reduce(getitem, keys, data)
|
except KeyError:
|
raise ValidationError('"{data_key}" key is expected in task data'.format(data_key=data_key))
|
else:
|
if data_key not in data:
|
raise ValidationError('"{data_key}" key is expected in task data'.format(data_key=data_key))
|
data_item = data[data_key]
|
|
if is_array:
|
expected_types = (list,)
|
else:
|
expected_types = _DATA_TYPES.get(data_type, (str,))
|
|
if not isinstance(data_item, tuple(expected_types)):
|
raise ValidationError(
|
"data['{data_key}']={data_value} is of type '{type}', "
|
'but the object tag {data_type} expects the following types: {expected_types}'.format(
|
data_key=data_key,
|
data_value=data_item,
|
type=type(data_item).__name__,
|
data_type=data_type,
|
expected_types=[e.__name__ for e in expected_types],
|
)
|
)
|
|
return data
|
|
@staticmethod
|
def check_data_and_root(project, data, dict_is_root=False):
|
"""Check data consistent and data is dict with task or dict['task'] is task
|
|
:param project:
|
:param data:
|
:param dict_is_root:
|
:return:
|
"""
|
try:
|
TaskValidator.check_data(project, data)
|
except ValidationError as e:
|
if dict_is_root:
|
raise ValidationError(e.detail[0] + ' [assume: item as is = task root with values] ')
|
else:
|
raise ValidationError(e.detail[0] + ' [assume: item["data"] = task root with values]')
|
|
@staticmethod
|
def check_allowed(task):
|
# task is required
|
if 'data' not in task:
|
return False
|
|
# everything is ok
|
return True
|
|
@staticmethod
|
def raise_if_wrong_class(task, key, class_def):
|
if key in task and not isinstance(task[key], class_def):
|
if isinstance(class_def, tuple):
|
class_def = ' or '.join([c.__name__ for c in class_def])
|
else:
|
class_def = class_def.__name__
|
raise ValidationError('Task[{key}] must be {class_def}'.format(key=key, class_def=class_def))
|
|
def validate(self, task):
|
"""Validate whole task with task['data'] and task['annotations']. task['predictions']"""
|
# task is class
|
if hasattr(task, 'data'):
|
self.check_data_and_root(self.project, task.data)
|
return task
|
|
# self.instance is loaded by get_object of view
|
if self.instance and hasattr(self.instance, 'data'):
|
if isinstance(self.instance.data, dict):
|
data = self.instance.data
|
elif isinstance(self.instance.data, str):
|
try:
|
data = json.loads(self.instance.data)
|
except ValueError as e:
|
raise ValidationError("Can't parse task data: " + str(e))
|
else:
|
raise ValidationError(
|
'Field "data" must be string or dict, but not "' + type(self.instance.data) + '"'
|
)
|
self.check_data_and_root(self.instance.project, data)
|
return task
|
|
# check task is dict
|
if not isinstance(task, dict):
|
raise ValidationError('Task root must be dict with "data", "meta", "annotations", "predictions" fields')
|
|
# task[data] | task[annotations] | task[predictions] | task[meta]
|
if self.check_allowed(task):
|
# task[data]
|
self.raise_if_wrong_class(task, 'data', (dict, list))
|
self.check_data_and_root(self.project, task['data'])
|
|
# task[annotations]: we can't use AnnotationSerializer for validation
|
# because it's much different with validation we need here
|
self.raise_if_wrong_class(task, 'annotations', list)
|
for annotation in task.get('annotations', []):
|
if not isinstance(annotation, dict):
|
logger.warning('Annotation must be dict, but "%s" found', str(type(annotation)))
|
continue
|
|
ok = 'result' in annotation
|
if not ok:
|
raise ValidationError('Annotation must have "result" fields')
|
|
# check result is list
|
if not isinstance(annotation.get('result', []), list):
|
raise ValidationError('"result" field in annotation must be list')
|
|
# task[predictions]
|
self.raise_if_wrong_class(task, 'predictions', list)
|
for prediction in task.get('predictions', []):
|
if not isinstance(prediction, dict):
|
logger.warning('Prediction must be dict, but "%s" found', str(type(prediction)))
|
continue
|
|
ok = 'result' in prediction
|
if not ok:
|
raise ValidationError('Prediction must have "result" fields')
|
|
# task[meta]
|
self.raise_if_wrong_class(task, 'meta', (dict, list))
|
|
# task is data as is, validate task as data and move it to task['data']
|
else:
|
self.check_data_and_root(self.project, task, dict_is_root=True)
|
task = {'data': task}
|
|
return task
|
|
@staticmethod
|
def format_error(i, detail, item):
|
if len(detail) == 1:
|
code = (str(detail[0].code + ' ')) if detail[0].code != 'invalid' else ''
|
return 'Error {code} at item {i}: {detail} :: {item}'.format(code=code, i=i, detail=detail[0], item=item)
|
else:
|
errors = ', '.join(detail)
|
codes = str([d.code for d in detail])
|
return 'Errors {codes} at item {i}: {errors} :: {item}'.format(codes=codes, i=i, errors=errors, item=item)
|
|
def to_internal_value(self, data):
|
"""Body of run_validation for all data items"""
|
if data is None:
|
raise ValidationError('All tasks are empty (None)')
|
|
if not isinstance(data, list):
|
raise ValidationError('data is not a list')
|
|
if len(data) == 0:
|
raise ValidationError('data is empty')
|
|
ret, errors = [], []
|
self.annotation_count, self.prediction_count = 0, 0
|
for i, item in enumerate(data):
|
try:
|
validated = self.validate(item)
|
except ValidationError as exc:
|
error = self.format_error(i, exc.detail, item)
|
errors.append(error)
|
# do not print to user too many errors
|
if len(errors) >= 100:
|
errors[99] = '...'
|
break
|
else:
|
ret.append(validated)
|
errors.append({})
|
|
if 'annotations' in item:
|
self.annotation_count += len(item['annotations'])
|
if 'predictions' in item:
|
self.prediction_count += len(item['predictions'])
|
|
if any(errors):
|
logger.warning("Can't deserialize tasks due to " + str(errors))
|
raise ValidationError(errors)
|
|
return ret
|
|
|
def is_url(string):
|
try:
|
result = urlparse(string.strip())
|
return all([result.scheme, result.netloc])
|
except ValueError:
|
return False
|
