from collections import namedtuple
import traceback
import logging
import uuid
import attr
import os
from .. import repositories, entities, exceptions
from ..services.api_client import ApiClient
logger = logging.getLogger(name='dtlpy')
[docs]@attr.s
class Recipe(entities.BaseEntity):
"""
Recipe object
"""
id = attr.ib()
creator = attr.ib()
url = attr.ib(repr=False)
title = attr.ib()
project_ids = attr.ib()
description = attr.ib()
ontology_ids = attr.ib(repr=False)
instructions = attr.ib(repr=False)
examples = attr.ib(repr=False)
custom_actions = attr.ib(repr=False)
metadata = attr.ib()
created_at = attr.ib()
updated_at = attr.ib()
updated_by = attr.ib()
# name change
ui_settings = attr.ib()
# platform
_client_api = attr.ib(type=ApiClient, repr=False)
# entities
_dataset = attr.ib(repr=False, default=None)
_project = attr.ib(repr=False, default=None)
# repositories
_repositories = attr.ib(repr=False)
@property
def customActions(self):
return self.custom_actions
@property
def ontologyIds(self):
logger.warning('Deprecation Warning - ontologyIds will be Deprecation from version 1.52.0 use ontology_ids')
return self.ontology_ids
[docs] @classmethod
def from_json(cls, _json, client_api, dataset=None, project=None, is_fetched=True):
"""
Build a Recipe entity object from a json
:param dict _json: _json response from host
:param dtlpy.entities.dataset.Dataset dataset: Dataset entity
:param dtlpy.entities.project.Project project: project entity
:param dl.ApiClient client_api: ApiClient entity
:param bool is_fetched: is Entity fetched from Platform
:return: Recipe object
"""
project_ids = _json.get('projectIds', None)
if project is not None and project_ids is not None:
if project.id not in project_ids:
logger.warning('Recipe has been fetched from a project that is not belong to it')
project = None
inst = cls(
client_api=client_api,
dataset=dataset,
project=project,
id=_json['id'],
creator=_json.get('creator', None),
url=_json.get('url', None),
title=_json.get('title', None),
project_ids=project_ids,
description=_json.get('description', None),
ontology_ids=_json.get('ontologyIds', None),
instructions=_json.get('instructions', None),
ui_settings=_json.get('uiSettings', None),
metadata=_json.get('metadata', None),
examples=_json.get('examples', None),
custom_actions=_json.get('customActions', None),
created_at=_json.get("createdAt", None),
updated_at=_json.get("updatedAt", None),
updated_by=_json.get("updatedBy", None))
inst.is_fetched = is_fetched
return inst
@staticmethod
def _protected_from_json(_json, client_api, project=None, dataset=None, is_fetched=True):
"""
Same as from_json but with try-except to catch if error
:param _json: platform json
:param client_api: ApiClient entity
:param project: project entity
:param dataset: dataset entity
:param is_fetched: is Entity fetched from Platform
:return:
"""
try:
recipe = Recipe.from_json(_json=_json,
client_api=client_api,
project=project,
dataset=dataset,
is_fetched=is_fetched)
status = True
except Exception:
recipe = traceback.format_exc()
status = False
return status, recipe
@_repositories.default
def set_repositories(self):
reps = namedtuple('repositories',
field_names=['ontologies', 'recipes'])
if self._dataset is None and self._project is None:
recipes = repositories.Recipes(client_api=self._client_api, dataset=self._dataset, project=self._project)
elif self._dataset is not None:
recipes = self.dataset.recipes
else:
recipes = self.project.recipes
r = reps(ontologies=repositories.Ontologies(recipe=self, client_api=self._client_api),
recipes=recipes)
return r
@property
def dataset(self):
if self._dataset is not None:
assert isinstance(self._dataset, entities.Dataset)
return self._dataset
@property
def project(self):
if self._project is not None:
assert isinstance(self._project, entities.Project)
return self._project
@property
def recipes(self):
assert isinstance(self._repositories.recipes, repositories.Recipes)
return self._repositories.recipes
@property
def ontologies(self):
assert isinstance(self._repositories.ontologies, repositories.Ontologies)
return self._repositories.ontologies
[docs] def to_json(self):
"""
Returns platform _json format of object
:return: platform json format of object
:rtype: dict
"""
_json = attr.asdict(self, filter=attr.filters.exclude(attr.fields(Recipe)._client_api,
attr.fields(Recipe)._dataset,
attr.fields(Recipe)._project,
attr.fields(Recipe).project_ids,
attr.fields(Recipe).ui_settings,
attr.fields(Recipe)._repositories,
attr.fields(Recipe).custom_actions,
attr.fields(Recipe).ontology_ids,
attr.fields(Recipe).created_at,
attr.fields(Recipe).updated_at,
attr.fields(Recipe).updated_by,
))
_json['uiSettings'] = self.ui_settings
_json['projectIds'] = self.project_ids
_json['customActions'] = self.custom_actions
_json['ontologyIds'] = self.ontology_ids
_json['createdAt'] = self.created_at
_json['updatedAt'] = self.updated_at
_json['updatedBy'] = self.updated_by
return _json
@property
def platform_url(self):
return self._client_api._get_resource_url("projects/{}/recipes/{}".format(self.project_ids[0], self.id))
[docs] def open_in_web(self):
"""
Open the recipes in web platform
:return:
"""
self._client_api._open_in_web(url=self.platform_url)
[docs] def delete(self, force: bool = False):
"""
Delete recipe from platform
:param bool force: force delete recipe
:return: True
:rtype: bool
"""
return self.recipes.delete(self.id, force=force)
[docs] def update(self, system_metadata=False):
"""
Update Recipe
:param bool system_metadata: bool - True, if you want to change metadata system
:return: Recipe object
:rtype: dtlpy.entities.recipe.Recipe
"""
return self.recipes.update(recipe=self, system_metadata=system_metadata)
[docs] def clone(self, shallow=False):
"""
Clone Recipe
:param bool shallow: If True, link ot existing ontology, clones all ontology that are link to the recipe as well
:return: Cloned ontology object
:rtype: dtlpy.entities.recipe.Recipe
"""
return self.recipes.clone(recipe=self,
shallow=shallow)
[docs] def get_annotation_template_id(self, template_name):
"""
Get annotation template id by template name
:param str template_name:
:return: template id or None if does not exist
"""
collection_templates = list()
if 'system' in self.metadata and 'collectionTemplates' in self.metadata['system']:
collection_templates = self.metadata['system']['collectionTemplates']
for template in collection_templates:
if "name" and 'id' in template:
if template_name == template['name']:
return template['id']
raise exceptions.NotFound('404', "annotation template {!r} not found".format(template_name))
[docs] def add_instruction(self, annotation_instruction_file):
"""
Add instruction to recipe
:param str annotation_instruction_file: file path or url of the recipe instruction
"""
_, ext = os.path.splitext(annotation_instruction_file)
if ext != '.pdf':
raise exceptions.PlatformException(error='400',
message='file Must be pdf')
for project_id in self.project_ids:
project = repositories.Projects(client_api=self._client_api).get(project_id=project_id)
dataset = project.datasets._get_binaries_dataset()
remote_path = '/.dataloop/recipes/{}/instructions'.format(self.id)
instruction_item = dataset.items.upload(local_path=annotation_instruction_file,
remote_path=remote_path,
remote_name=str(uuid.uuid4()) + '.pdf',
overwrite=True)
self.metadata['system']['instructionDocument'] = {'itemId': instruction_item.id,
'datasetId': dataset.id,
'name': instruction_item.name}
self.update(True)
[docs] def upload_annotations_verification_file(self, local_path: str, overwrite: bool = False) -> entities.Item:
"""
Add Annotations Verification js file to the recipe.
:param str local_path: file path of the annotations verification js file.
:param bool overwrite: overwrite exiting file if the local and the remote names are matching
:return: annotations verification js item.
"""
validation_file_metadata = self.metadata.get("system", dict()).get("validationFile", None)
if validation_file_metadata is None:
validation_file_metadata = dict()
remote_name = validation_file_metadata.get("name", None)
local_name = os.path.basename(local_path)
binaries_dataset = self._project.datasets._get_binaries_dataset()
remote_path = f"/.dataloop/recipes/{self.id}/verification/"
if remote_name is None or overwrite or remote_name != local_name:
validation_item = binaries_dataset.items.upload(
local_path=local_path,
remote_path=remote_path,
remote_name=local_name,
overwrite=True
)
self.metadata["system"]["validationFile"] = {
"itemId": validation_item.id,
"datasetId": binaries_dataset.id,
"name": local_name
}
self.update(system_metadata=True)
else:
logger.debug(f"Existing Annotations Validation Script was found.")
validation_item_id = self.metadata["system"]["validationFile"]["itemId"]
validation_item = binaries_dataset.items.get(item_id=validation_item_id)
return validation_item