Source code for dtlpy.entities.ontology

import os
import traceback
from collections import namedtuple
import logging
import random
import uuid
import attr

from .. import entities, PlatformException, repositories, services, exceptions

from .label import Label

logger = logging.getLogger(name='dtlpy')


class AttributesTypes:
    CHECKBOX = "checkbox"
    RADIO_BUTTON = "radio_button"
    SLIDER = "range"
    YES_NO = "boolean"
    FREE_TEXT = "freeText"


class AttributesRange:
    def __init__(self, min_range, max_range, step):
        self.min_range = min_range
        self.max_range = max_range
        self.step = step

    def to_json(self):
        return {'min': self.min_range, 'max': self.max_range, 'step': self.step}


class LabelHandlerMode:
    ADD = "add"
    UPDATE = "update"
    UPSERT = "upsert"


[docs]@attr.s class Ontology(entities.BaseEntity): """ Ontology object """ # api _client_api = attr.ib(type=services.ApiClient, repr=False) # params id = attr.ib() creator = attr.ib() url = attr.ib(repr=False) title = attr.ib() labels = attr.ib(repr=False) metadata = attr.ib(repr=False) attributes = attr.ib() # entities _recipe = attr.ib(repr=False, default=None) _dataset = attr.ib(repr=False, default=None) _project = attr.ib(repr=False, default=None) # repositories _repositories = attr.ib(repr=False) # defaults _instance_map = attr.ib(default=None, repr=False) _color_map = attr.ib(default=None, repr=False) @_repositories.default def set_repositories(self): reps = namedtuple('repositories', field_names=['ontologies', 'datasets', 'projects']) if self._recipe is None: ontologies = repositories.Ontologies(client_api=self._client_api, recipe=self._recipe) else: ontologies = self.recipe.ontologies r = reps(ontologies=ontologies, datasets=repositories.Datasets(client_api=self._client_api), projects=repositories.Projects(client_api=self._client_api)) return r @property def recipe(self): if self._recipe is not None: assert isinstance(self._recipe, entities.Recipe) return self._recipe @property def dataset(self): if self._dataset is None: if self.recipe is not None: self._dataset = self.recipe.dataset if self._dataset is not None: assert isinstance(self._dataset, entities.Dataset) return self._dataset @property def project(self): if self._project is None: if 'system' in self.metadata: project_id = self.metadata['system'].get('projectIds', None) if project_id is not None: self._project = self.projects.get(project_id=project_id[0]) elif self.dataset is not None: self._project = self.dataset.project if self._project is not None: assert isinstance(self._project, entities.Project) return self._project @property def ontologies(self): if self._repositories.ontologies is not None: assert isinstance(self._repositories.ontologies, repositories.Ontologies) return self._repositories.ontologies @property def projects(self): if self._repositories.projects is not None: assert isinstance(self._repositories.projects, repositories.Projects) return self._repositories.projects @property def labels_flat_dict(self): flatten_dict = dict() def add_to_dict(tag, father): flatten_dict[tag] = father for child in father.children: add_to_dict('{}.{}'.format(tag, child.tag), child) for label in self.labels: add_to_dict(label.tag, label) return flatten_dict @property def instance_map(self): """ instance mapping for creating instance mask :return dictionary {label: map_id} :rtype: dict """ if self._instance_map is None: labels = [label for label in self.labels_flat_dict] labels.sort() # each label gets index as instance id self._instance_map = {label: (i_label + 1) for i_label, label in enumerate(labels)} return self._instance_map @instance_map.setter def instance_map(self, value: dict): """ instance mapping for creating instance mask :param value: dictionary {label: map_id} :rtype: dict """ if not isinstance(value, dict): raise ValueError('input must be a dictionary of {label_name: instance_id}') self._instance_map = value @property def color_map(self): """ Color mapping of labels, {label: rgb} :return: dict :rtype: dict """ if self._color_map is None: self._color_map = {k: v.rgb for k, v in self.labels_flat_dict.items()} return self._color_map @color_map.setter def color_map(self, values): """ Color mapping of labels, {label: rgb} :param values: dict {label: rgb} :return: """ if not isinstance(values, dict): raise ValueError('input must be a dict. got: {}'.format(type(values))) self._color_map = values @staticmethod def _protected_from_json(_json, client_api, recipe, dataset, project, is_fetched=True): """ Same as from_json but with try-except to catch if error :param _json: platform json :param client_api: ApiClient entity :return: """ try: ontology = Ontology.from_json(_json=_json, client_api=client_api, project=project, dataset=dataset, recipe=recipe, is_fetched=is_fetched) status = True except Exception: ontology = traceback.format_exc() status = False return status, ontology @property def _use_attributes_2(self): return os.environ.get("USE_ATTRIBUTE_2", 'false') == 'true'
[docs] @classmethod def from_json(cls, _json, client_api, recipe, dataset=None, project=None, is_fetched=True): """ Build an Ontology entity object from a json :param bool is_fetched: is Entity fetched from Platform :param dtlpy.entities.project.Project project: project entity :param dtlpy.entities.dataset.Dataset dataset: dataset :param dict _json: _json response from host :param dtlpy.entities.recipe.Recipe recipe: ontology's recipe :param dl.ApiClient client_api: ApiClient entity :return: Ontology object :rtype: dtlpy.entities.ontology.Ontology """ if not os.environ.get("USE_ATTRIBUTE_2", 'false') == 'true': attributes = _json.get("attributes", []) else: attributes = _json.get('metadata', {}).get("attributes", []) labels = list() for root in _json["roots"]: labels.append(entities.Label.from_root(root=root)) inst = cls( metadata=_json.get("metadata", None), creator=_json.get("creator", None), url=_json.get("url", None), id=_json["id"], title=_json.get("title", None), attributes=attributes, client_api=client_api, project=project, dataset=dataset, recipe=recipe, labels=labels, ) inst.is_fetched = is_fetched return inst
[docs] def to_json(self): """ Returns platform _json format of object :return: platform json format of object :rtype: dict """ roots = [label.to_root() for label in self.labels] _json = attr.asdict(self, filter=attr.filters.exclude(attr.fields(Ontology)._client_api, attr.fields(Ontology)._recipe, attr.fields(Ontology)._project, attr.fields(Ontology)._dataset, attr.fields(Ontology)._instance_map, attr.fields(Ontology)._color_map, attr.fields(Ontology)._repositories)) _json["roots"] = roots return _json
[docs] def delete(self): """ Delete recipe from platform :return: True """ return self.ontologies.delete(self.id)
[docs] def update(self, system_metadata=False): """ Update items metadata :param bool system_metadata: bool - True, if you want to change metadata system :return: Ontology object """ return self.ontologies.update(self, system_metadata=system_metadata)
def _add_children(self, label_name, children, labels_node, mode): for child in children: if not isinstance(child, entities.Label): if isinstance(child, dict): if "label_name" in child: child = dict(child) child["label_name"] = "{}.{}".format(label_name, child["label_name"]) labels_node += self._base_labels_handler(labels=[child], update_ontology=False, mode=mode) else: raise PlatformException("400", "Invalid parameters - child list must have label name attribute") else: raise PlatformException("400", "Invalid parameters - child must be a dict type") else: child.tag = "{}.{}".format(label_name, child.tag) labels_node += self._base_labels_handler(labels=child, update_ontology=False, mode=mode) return labels_node def _labels_handler_update_mode(self, json_req, upsert=False, log_error=True): json_req['upsert'] = upsert success, response = self._client_api.gen_request(req_type="PATCH", path="/ontologies/%s/labels" % self.id, json_req=json_req, log_error=log_error) if success: logger.debug("Labels {} has been added successfully".format(json_req)) else: raise exceptions.PlatformException(response) return response def _labels_handler_add_mode(self, json_req): success, response = self._client_api.gen_request(req_type="PATCH", path="/ontologies/%s/addLabels" % self.id, json_req=json_req) if success: logger.debug("Labels {} has been added successfully".format(json_req)) else: raise exceptions.PlatformException(response) return response def _base_labels_handler(self, labels, update_ontology=True, mode=LabelHandlerMode.UPSERT): """ Add a single label to ontology using add label endpoint , nested label is also supported :param labels = list of labels :param update_ontology - return json_req if False :param mode add, update or upsert, relevant on update_ontology=True only :return: Ontology updated entire label entity """ labels_node = list() if mode not in [LabelHandlerMode.ADD, LabelHandlerMode.UPDATE, LabelHandlerMode.UPSERT]: raise ValueError('mode must be on of: "add", "update", "upsert"') if not isinstance(labels, list): # for case that add label get one label labels = [labels] for label in labels: if isinstance(label, str): # Generate label from string label = entities.Label(tag=label) elif isinstance(label, dict): # Generate label from dict label = Label.from_root(label) elif isinstance(label, entities.Label): ... else: raise ValueError( 'Unsupported type for `labels`. Expected a list of (str, dict, dl.Label). Got: {}'.format( type(label))) # label entity label_node = {"tag": label.tag} if label.color is not None: label_node["color"] = label.hex if label.attributes is not None: label_node["attributes"] = label.attributes if label.display_label is not None: label_node["displayLabel"] = label.display_label if label.display_data is not None: label_node["displayData"] = label.display_data labels_node.append(label_node) children = label.children self._add_children(label.tag, children, labels_node, mode=mode) if not update_ontology or not len(labels_node): return labels_node json_req = { "labelsNode": labels_node } if mode == LabelHandlerMode.UPDATE: response = self._labels_handler_update_mode(json_req) else: response = self._labels_handler_update_mode(json_req, upsert=True, log_error=False) added_label = list() if "roots" not in response.json(): raise exceptions.PlatformException("error fetching updated labels from server") for root in response.json()["roots"]: # to get all labels added_label.append(entities.Label.from_root(root=root)) self.labels = added_label return added_label def _add_image_label(self, icon_path): display_data = dict() if self.project is not None: dataset = self.project.datasets.get(dataset_name='Binaries') elif self.dataset is not None: dataset = self.dataset.project.datasets.get(dataset_name='Binaries') else: raise ValueError('must have project or dataset to create with icon path') platform_path = "/.dataloop/ontologies/{}/labelDisplayImages/".format(self.id) basename = os.path.basename(icon_path) item = dataset.items.upload(local_path=icon_path, remote_path=platform_path, remote_name='{}-{}'.format(uuid.uuid4().hex, basename)) display_data['displayImage'] = dict() display_data['displayImage']['itemId'] = item.id display_data['displayImage']['datasetId'] = item.dataset_id return display_data def _label_handler(self, label_name, color=None, children=None, attributes=None, display_label=None, label=None, add=True, icon_path=None, update_ontology=False, mode=LabelHandlerMode.UPSERT): """ Add a single label to ontology :param label_name: label name :param color: optional - if not given a random color will be selected :param children: optional - children :param attributes: optional - attributes :param display_label: optional - display_label :param label: label :param add:to add or not :param icon_path: path to image to be display on label :param update_ontology: update the ontology, default = False for backward compatible :param mode add, update or upsert, relevant on update_ontology=True only :return: Label entity """ if update_ontology: if isinstance(label, entities.Label) or isinstance(label, str): return self._base_labels_handler(labels=label, update_ontology=update_ontology, mode=mode) else: display_data = dict() if icon_path is not None: display_data = self._add_image_label(icon_path=icon_path) return self._base_labels_handler(labels={"tag": label_name, "displayLabel": display_label, "color": color, "attributes": attributes, "children": children, "displayData": display_data }, update_ontology=update_ontology, mode=mode) if not isinstance(label, entities.Label): if "." in label_name: raise PlatformException("400", "Invalid parameters - nested label can work with update_ontology option only") if attributes is None: attributes = list() if not isinstance(attributes, list): attributes = [attributes] # get random color if none given if color is None: color = (random.randint(0, 255), random.randint(0, 255), random.randint(0, 255)) if children is None: children = list() if not isinstance(children, list): children = [children] # add children added_children = list() for child in children: if not isinstance(child, entities.Label): added_children.append(self._label_handler(**child, add=False)) else: added_children.append(child) if display_label is None: display_label = "" if len(label_name.split("_")) == 1: display_label = label_name[0].upper() + label_name[1:] else: for word in label_name.split("_"): display_label += word[0].upper() + word[1:] + " " display_label = display_label[0:-1] display_data = dict() if icon_path is not None: display_data = self._add_image_label(icon_path=icon_path) root = { "value": { "tag": label_name, "displayLabel": display_label, "color": color, "attributes": attributes, "displayData": display_data }, "children": list(), } added_label = entities.Label.from_root(root) added_label.children = added_children else: added_label = label if add and self._validate_label(added_label=added_label, mode=mode, color=color, children=children, attributes=attributes, display_label=display_label, display_data=icon_path): self.labels.append(added_label) self._base_labels_handler(labels=added_label, update_ontology=True, mode=mode) return added_label def _validate_label(self, added_label, mode=LabelHandlerMode.UPSERT, color=None, children=None, attributes=None, display_label=None, display_data=None): """ check if the label is exist """ for i in range(len(self.labels)): if self.labels[i].tag == added_label.tag: if mode == LabelHandlerMode.UPDATE: if color: self.labels[i].color = added_label.color if children: self.labels[i].children = added_label.children if attributes: self.labels[i].attributes = added_label.attributes if display_label: self.labels[i].display_label = added_label.display_label if display_data: self.labels[i].display_data = added_label.display_data return False return True def _labels_handler(self, label_list, update_ontology=False, mode=LabelHandlerMode.UPSERT): """ Adds a list of labels to ontology :param list label_list: a list of labels to add to the dataset's ontology. each value should be a dict, dl.Label or a string if dictionary, should look like this: {"value": {"tag": "name of the label", "displayLabel": "display name on the platform", "color": "#hex value", "attributes": [attributes]}, "children": [children]} :param update_ontology: update the ontology, default = False for backward compatible :param mode add, update or upsert, relevant on update_ontology=True only :return: List of label entities added """ if update_ontology: return self._base_labels_handler(labels=label_list, mode=mode) labels = list() for label in label_list: if isinstance(label, str): label = entities.Label(tag=label) if isinstance(label, entities.Label): # label entity labels.append(label) else: # dictionary labels.append(Label.from_root(label)) added_labels = list() for label in labels: added_labels.append(self._label_handler(label.tag, label=label, update_ontology=update_ontology)) return added_labels
[docs] def delete_labels(self, label_names): """ Delete labels from ontology :param label_names: label object/ label name / list of label objects / list of label names :return: """ if not isinstance(label_names, list): label_names = [label_names] if isinstance(label_names[0], entities.Label): label_names = [label.tag for label in label_names] for label in label_names: self.__delete_label(label) self.update()
def __delete_label(self, label_name): if label_name in self.instance_map.keys(): labels = self.labels label_chain = label_name.split('.') while len(label_chain) > 1: label_name = label_chain.pop(0) for i_label, label in enumerate(labels): if label.tag == label_name: labels = labels[i_label].children break label_name = label_chain[0] for i_label, label in enumerate(labels): if label.tag == label_name: labels.pop(i_label)
[docs] def add_label(self, label_name, color=None, children=None, attributes=None, display_label=None, label=None, add=True, icon_path=None, update_ontology=False): """ Add a single label to ontology :param str label_name: str - label name :param tuple color: color :param children: children (sub labels) :param list attributes: attributes :param str display_label: display_label :param dtlpy.entities.label.Label label: label :param bool add: to add or not :param str icon_path: path to image to be display on label :param bool update_ontology: update the ontology, default = False for backward compatible :return: Label entity :rtype: dtlpy.entities.label.Label **Example**: .. code-block:: python ontology.add_label(label_name='person', color=(34, 6, 231), attributes=['big', 'small']) """ return self._label_handler(label_name=label_name, color=color, children=children, attributes=attributes, display_label=display_label, label=label, add=add, icon_path=icon_path, update_ontology=update_ontology)
[docs] def add_labels(self, label_list, update_ontology=False): """ Adds a list of labels to ontology :param list label_list: list of labels [{"value": {"tag": "tag", "displayLabel": "displayLabel", "color": "#color", "attributes": [attributes]}, "children": [children]}] :param bool update_ontology: update the ontology, default = False for backward compatible :return: List of label entities added **Example**: .. code-block:: python ontology.add_labels(label_list=label_list) """ self._labels_handler(label_list=label_list, update_ontology=update_ontology, mode=LabelHandlerMode.UPSERT)
[docs] def update_label(self, label_name, color=None, children=None, attributes=None, display_label=None, label=None, add=True, icon_path=None, upsert=False, update_ontology=False): """ Update a single label to ontology :param str label_name: str - label name :param tuple color: color :param children: children (sub labels) :param list attributes: attributes :param str display_label: display_label :param dtlpy.entities.label.Label label: label :param bool add: to add or not :param str icon_path: path to image to be display on label :param bool update_ontology: update the ontology, default = False for backward compatible :param bool upsert: if True will add in case it does not existing :return: Label entity :rtype: dtlpy.entities.label.Label **Example**: .. code-block:: python ontology.update_label(label_name='person', color=(34, 6, 231), attributes=['big', 'small']) """ if upsert: mode = LabelHandlerMode.UPSERT else: mode = LabelHandlerMode.UPDATE return self._label_handler(label_name=label_name, color=color, children=children, attributes=attributes, display_label=display_label, label=label, add=add, icon_path=icon_path, update_ontology=update_ontology, mode=mode)
[docs] def update_labels(self, label_list, upsert=False, update_ontology=False): """ Update a list of labels to ontology :param list label_list: list of labels [{"value": {"tag": "tag", "displayLabel": "displayLabel", "color": "#color", "attributes": [attributes]}, "children": [children]}] :param bool upsert: if True will add in case it does not existing :param bool update_ontology: update the ontology, default = False for backward compatible :return: List of label entities added **Example**: .. code-block:: python ontology.update_labels(label_list=label_list) """ if upsert: mode = LabelHandlerMode.UPSERT else: mode = LabelHandlerMode.UPDATE self._labels_handler(label_list=label_list, update_ontology=update_ontology, mode=mode)
[docs] def update_attributes(self, title: str, key: str, attribute_type, scope: list = None, optional: bool = None, values: list = None, attribute_range=None): """ ADD a new attribute or update if exist :param str title: attribute title :param str key: the key of the attribute must br unique :param AttributesTypes attribute_type: dl.AttributesTypes your attribute type :param list scope: list of the labels or * for all labels :param bool optional: optional attribute :param list values: list of the attribute values ( for checkbox and radio button) :param dict or AttributesRange attribute_range: dl.AttributesRange object :return: true in success :rtype: bool """ return self.ontologies.update_attributes( ontology_id=self.id, title=title, key=key, attribute_type=attribute_type, scope=scope, optional=optional, values=values, attribute_range=attribute_range)
[docs] def delete_attributes(self, keys: list): """ Delete a bulk of attributes :param list keys: Keys of attributes to delete :return: True if success :rtype: bool **Example**: .. code-block:: python ontology.delete_attributes(['1']) """ return self.ontologies.delete_attributes(ontology_id=self.id, keys=keys)