# coding: utf-8
"""download/upload/edit :class:`Annotation<supervisely.annotation.annotation.Annotation>`"""
# docs
from __future__ import annotations
import json
from collections import defaultdict
from copy import deepcopy
from typing import Any, Callable, Dict, List, NamedTuple, Optional, Union
from tqdm import tqdm
from supervisely._utils import batched
from supervisely.annotation.annotation import Annotation, AnnotationJsonFields
from supervisely.annotation.label import Label, LabelJsonFields
from supervisely.api.module_api import ApiField, ModuleApi
from supervisely.geometry.alpha_mask import AlphaMask
from supervisely.geometry.constants import BITMAP
from supervisely.project.project_meta import ProjectMeta
class AnnotationInfo(NamedTuple):
"""
AnnotationInfo
"""
image_id: int
image_name: str
annotation: dict
created_at: str
updated_at: str
def to_json(self) -> Dict[str, Any]:
"""
Convert AnnotationInfo to JSON format.
:return: AnnotationInfo in JSON format.
:rtype: :class:`Dict[str, Any]`
"""
return {
ApiField.IMAGE_ID: self.image_id,
ApiField.IMAGE_NAME: self.image_name,
ApiField.ANNOTATION: self.annotation,
ApiField.CREATED_AT: self.created_at,
ApiField.UPDATED_AT: self.updated_at,
}
[docs]class AnnotationApi(ModuleApi):
"""
Annotation for a single image. :class:`AnnotationApi<AnnotationApi>` object is immutable.
:param api: API connection to the server.
:type api: Api
:Usage example:
.. code-block:: python
import os
from dotenv import load_dotenv
import supervisely as sly
# Load secrets and create API object from .env file (recommended)
# Learn more here: https://developer.supervisely.com/getting-started/basics-of-authentication
if sly.is_development():
load_dotenv(os.path.expanduser("~/supervisely.env"))
api = sly.Api.from_env()
# Pass values into the API constructor (optional, not recommended)
# api = sly.Api(server_address="https://app.supervise.ly", token="4r47N...xaTatb")
dataset_id = 254737
ann_infos = api.annotation.get_list(dataset_id)
"""
[docs] @staticmethod
def info_sequence():
"""
NamedTuple AnnotationInfo information about Annotation.
:Example:
.. code-block:: python
AnnotationInfo(image_id=121236919,
image_name='IMG_1836',
annotation={'description': '', 'tags': [], 'size': {'height': 800, 'width': 1067}, 'objects': []},
created_at='2019-12-19T12:06:59.435Z',
updated_at='2021-02-06T11:07:26.080Z')
"""
return [
ApiField.IMAGE_ID,
ApiField.IMAGE_NAME,
ApiField.ANNOTATION,
ApiField.CREATED_AT,
ApiField.UPDATED_AT,
]
[docs] @staticmethod
def info_tuple_name():
"""
NamedTuple name - **AnnotationInfo**.
"""
return "AnnotationInfo"
[docs] def get_list(
self,
dataset_id: int,
filters: Optional[List[Dict[str, str]]] = None,
progress_cb: Optional[Union[tqdm, Callable]] = None,
force_metadata_for_links: Optional[bool] = True,
) -> List[AnnotationInfo]:
"""
Get list of information about all annotations for a given dataset.
:param dataset_id: Dataset ID in Supervisely.
:type dataset_id: int
:param filters: List of parameters to sort output Annotations.
:type filters: List[dict], optional
:param progress_cb: Function for tracking download progress.
:type progress_cb: tqdm or callable, optional
:return: Information about Annotations. See :class:`info_sequence<info_sequence>`
:rtype: :class:`List[AnnotationInfo]`
:Usage example:
.. code-block:: python
import supervisely as sly
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
dataset_id = 254737
ann_infos = api.annotation.get_list(dataset_id)
print(json.dumps(ann_infos[0], indent=4))
# Output: [
# 121236918,
# "IMG_0748.jpeg",
# {
# "description": "",
# "tags": [],
# "size": {
# "height": 800,
# "width": 1067
# },
# "objects": []
# },
# "2019-12-19T12:06:59.435Z",
# "2021-02-06T11:07:26.080Z"
# ]
ann_infos_filter = api.annotation.get_list(dataset_id, filters={ 'field': 'name', 'operator': '=', 'value': 'IMG_1836' })
print(json.dumps(ann_infos_filter, indent=4))
# Output: [
# 121236919,
# "IMG_1836",
# {
# "description": "",
# "tags": [],
# "size": {
# "height": 800,
# "width": 1067
# },
# "objects": []
# },
# "2019-12-19T12:06:59.435Z",
# "2021-02-06T11:07:26.080Z"
# ]
"""
return self.get_list_all_pages(
"annotations.list",
{
ApiField.DATASET_ID: dataset_id,
ApiField.FILTER: filters or [],
ApiField.FORCE_METADATA_FOR_LINKS: force_metadata_for_links,
},
progress_cb,
)
[docs] def get_list_generator(
self,
dataset_id: int,
filters: Optional[List[Dict[str, str]]] = None,
progress_cb: Optional[Union[tqdm, Callable]] = None,
batch_size: Optional[int] = 50,
force_metadata_for_links: Optional[bool] = True,
) -> List[AnnotationInfo]:
"""
Get list of information about all annotations for a given dataset.
:param dataset_id: Dataset ID in Supervisely.
:type dataset_id: int
:param filters: List of parameters to sort output Annotations.
:type filters: List[dict], optional
:param progress_cb: Function for tracking download progress.
:type progress_cb: tqdm or callable, optional
:return: Information about Annotations. See :class:`info_sequence<info_sequence>`
:rtype: :class:`List[AnnotationInfo]`
:Usage example:
.. code-block:: python
import supervisely as sly
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
dataset_id = 254737
ann_infos = api.annotation.get_list(dataset_id)
print(json.dumps(ann_infos[0], indent=4))
# Output: [
# 121236918,
# "IMG_0748.jpeg",
# {
# "description": "",
# "tags": [],
# "size": {
# "height": 800,
# "width": 1067
# },
# "objects": []
# },
# "2019-12-19T12:06:59.435Z",
# "2021-02-06T11:07:26.080Z"
# ]
ann_infos_filter = api.annotation.get_list(dataset_id, filters={ 'field': 'name', 'operator': '=', 'value': 'IMG_1836' })
print(json.dumps(ann_infos_filter, indent=4))
# Output: [
# 121236919,
# "IMG_1836",
# {
# "description": "",
# "tags": [],
# "size": {
# "height": 800,
# "width": 1067
# },
# "objects": []
# },
# "2019-12-19T12:06:59.435Z",
# "2021-02-06T11:07:26.080Z"
# ]
"""
data = {
ApiField.DATASET_ID: dataset_id,
ApiField.FILTER: filters or [],
ApiField.FORCE_METADATA_FOR_LINKS: force_metadata_for_links,
ApiField.PAGINATION_MODE: ApiField.TOKEN,
}
if batch_size is not None:
data[ApiField.PER_PAGE] = batch_size
else:
# use default value on instance (learn in API documentation)
# 20k for instance
# 50 by default in SDK
pass
return self.get_list_all_pages_generator("annotations.list", data, progress_cb)
[docs] def download(
self,
image_id: int,
with_custom_data: Optional[bool] = False,
force_metadata_for_links: Optional[bool] = True,
) -> AnnotationInfo:
"""
Download AnnotationInfo by image ID from API.
:param image_id: Image ID in Supervisely.
:type image_id: int
:param with_custom_data: Include custom data in the response.
:type with_custom_data: bool, optional
:param force_metadata_for_links: Force metadata for links.
:type force_metadata_for_links: bool, optional
:return: Information about Annotation. See :class:`info_sequence<info_sequence>`
:rtype: :class:`AnnotationInfo`
:Usage example:
.. code-block:: python
import supervisely as sly
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
image_id = 121236918
ann_info = api.annotation.download(image_id)
print(json.dumps(ann_info, indent=4))
# Output: [
# 121236918,
# "IMG_0748.jpeg",
# {
# "description": "",
# "tags": [],
# "size": {
# "height": 800,
# "width": 1067
# },
# "objects": []
# },
# "2019-12-19T12:06:59.435Z",
# "2021-02-06T11:07:26.080Z"
# ]
"""
response = self._api.post(
"annotations.info",
{
ApiField.IMAGE_ID: image_id,
ApiField.WITH_CUSTOM_DATA: with_custom_data,
ApiField.FORCE_METADATA_FOR_LINKS: force_metadata_for_links,
ApiField.INTEGER_COORDS: False,
},
)
result = response.json()
# Convert annotation to pixel coordinate system
result[ApiField.ANNOTATION] = Annotation._to_pixel_coordinate_system_json(
result[ApiField.ANNOTATION]
)
# check if there are any AlphaMask geometries in the batch
additonal_geometries = defaultdict(int)
labels = result[ApiField.ANNOTATION][AnnotationJsonFields.LABELS]
for idx, label in enumerate(labels):
if label[LabelJsonFields.GEOMETRY_TYPE] == AlphaMask.geometry_name():
figure_id = label[LabelJsonFields.ID]
additonal_geometries[figure_id] = idx
# if so, download them separately and update the annotation
if len(additonal_geometries) > 0:
figure_ids = list(additonal_geometries.keys())
figures = self._api.image.figure.download_geometries_batch(figure_ids)
for figure_id, geometry in zip(figure_ids, figures):
label_idx = additonal_geometries[figure_id]
labels[label_idx].update({BITMAP: geometry})
ann_info = self._convert_json_info(result)
return ann_info
[docs] def download_json(
self,
image_id: int,
with_custom_data: Optional[bool] = False,
force_metadata_for_links: Optional[bool] = True,
) -> Dict[str, Union[str, int, list, dict]]:
"""
Download Annotation in json format by image ID from API.
:param image_id: Image ID in Supervisely.
:type image_id: int
:param with_custom_data: Include custom data in the response.
:type with_custom_data: bool, optional
:param force_metadata_for_links: Force metadata for links.
:type force_metadata_for_links: bool, optional
:return: Annotation in json format
:rtype: :class:`dict`
:Usage example:
.. code-block:: python
import supervisely as sly
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
image_id = 121236918
ann_json = api.annotation.download_json(image_id)
print(ann_json)
# Output: {
# "description": "",
# "tags": [],
# "size": {
# "height": 800,
# "width": 1067
# },
# "objects": []
# }
"""
return self.download(
image_id=image_id,
with_custom_data=with_custom_data,
force_metadata_for_links=force_metadata_for_links,
).annotation
[docs] def download_batch(
self,
dataset_id: int,
image_ids: List[int],
progress_cb: Optional[Union[tqdm, Callable]] = None,
with_custom_data: Optional[bool] = False,
force_metadata_for_links: Optional[bool] = True,
) -> List[AnnotationInfo]:
"""
Get list of AnnotationInfos for given dataset ID from API.
:param dataset_id: Dataset ID in Supervisely.
:type dataset_id: int
:param image_ids: List of integers.
:type image_ids: List[int]
:param progress_cb: Function for tracking download progress.
:type progress_cb: tqdm
:param with_custom_data: Include custom data in the response.
:type with_custom_data: bool, optional
:param force_metadata_for_links: Force metadata for links.
:type force_metadata_for_links: bool, optional
:return: Information about Annotations. See :class:`info_sequence<info_sequence>`
:rtype: :class:`List[AnnotationInfo]`
:Usage example:
.. code-block:: python
import supervisely as sly
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
dataset_id = 254737
image_ids = [121236918, 121236919]
p = tqdm(desc="Annotations downloaded: ", total=len(image_ids))
ann_infos = api.annotation.download_batch(dataset_id, image_ids, progress_cb=p)
# Output:
# {"message": "progress", "event_type": "EventType.PROGRESS", "subtask": "Annotations downloaded: ", "current": 0, "total": 2, "timestamp": "2021-03-16T15:20:06.168Z", "level": "info"}
# {"message": "progress", "event_type": "EventType.PROGRESS", "subtask": "Annotations downloaded: ", "current": 2, "total": 2, "timestamp": "2021-03-16T15:20:06.510Z", "level": "info"}
Optimizing the download process by using the context to avoid redundant API calls.:
# 1. Download the project meta
project_id = api.dataset.get_info_by_id(dataset_id).project_id
project_meta = api.project.get_meta(project_id)
# 2. Use the context to avoid redundant API calls
dataset_id = 254737
image_ids = [121236918, 121236919]
with sly.ApiContext(api, dataset_id=dataset_id, project_id=project_id, project_meta=project_meta):
ann_infos = api.annotation.download_batch(dataset_id, image_ids)
"""
# use context to avoid redundant API calls
context = self._api.optimization_context
context_dataset_id = context.get("dataset_id")
project_meta = context.get("project_meta")
project_id = context.get("project_id")
if dataset_id != context_dataset_id:
context["dataset_id"] = dataset_id
project_id, project_meta = None, None
if not isinstance(project_meta, ProjectMeta):
if project_id is None:
project_id = self._api.dataset.get_info_by_id(dataset_id).project_id
context["project_id"] = project_id
project_meta = ProjectMeta.from_json(self._api.project.get_meta(project_id))
context["project_meta"] = project_meta
need_download_alpha_masks = False
for obj_cls in project_meta.obj_classes:
if obj_cls.geometry_type == AlphaMask:
need_download_alpha_masks = True
break
id_to_ann = {}
for batch in batched(image_ids):
post_data = {
ApiField.DATASET_ID: dataset_id,
ApiField.IMAGE_IDS: batch,
ApiField.WITH_CUSTOM_DATA: with_custom_data,
ApiField.FORCE_METADATA_FOR_LINKS: force_metadata_for_links,
ApiField.INTEGER_COORDS: False,
}
results = self._api.post("annotations.bulk.info", data=post_data).json()
if need_download_alpha_masks is True:
additonal_geometries = defaultdict(tuple)
for ann_idx, ann_dict in enumerate(results):
# check if there are any AlphaMask geometries in the batch
for label_idx, label in enumerate(
ann_dict[ApiField.ANNOTATION][AnnotationJsonFields.LABELS]
):
if label[LabelJsonFields.GEOMETRY_TYPE] == AlphaMask.geometry_name():
figure_id = label[LabelJsonFields.ID]
additonal_geometries[figure_id] = (ann_idx, label_idx)
# if there are any AlphaMask geometries, download them separately and update the annotation
if len(additonal_geometries) > 0:
figure_ids = list(additonal_geometries.keys())
figures = self._api.image.figure.download_geometries_batch(figure_ids)
for figure_id, geometry in zip(figure_ids, figures):
ann_idx, label_idx = additonal_geometries[figure_id]
results[ann_idx][ApiField.ANNOTATION][AnnotationJsonFields.LABELS][
label_idx
].update({BITMAP: geometry})
for ann_dict in results:
# Convert annotation to pixel coordinate system
ann_dict[ApiField.ANNOTATION] = Annotation._to_pixel_coordinate_system_json(
ann_dict[ApiField.ANNOTATION]
)
ann_info = self._convert_json_info(ann_dict)
id_to_ann[ann_info.image_id] = ann_info
if progress_cb is not None:
progress_cb(len(batch))
ordered_results = [id_to_ann[image_id] for image_id in image_ids]
return ordered_results
[docs] def download_json_batch(
self,
dataset_id: int,
image_ids: List[int],
progress_cb: Optional[Union[tqdm, Callable]] = None,
force_metadata_for_links: Optional[bool] = True,
) -> List[Dict]:
"""
Get list of AnnotationInfos for given dataset ID from API.
:param dataset_id: Dataset ID in Supervisely.
:type dataset_id: int
:param image_ids: List of integers.
:type image_ids: List[int]
:param progress_cb: Function for tracking download progress.
:type progress_cb: tqdm
:param force_metadata_for_links: Force metadata for links.
:type force_metadata_for_links: bool, optional
:return: Information about Annotations. See :class:`info_sequence<info_sequence>`
:rtype: :class:`List[Dict]`
:Usage example:
.. code-block:: python
import supervisely as sly
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
dataset_id = 254737
image_ids = [121236918, 121236919]
p = tqdm(desc="Annotations downloaded: ", total=len(image_ids))
anns_jsons = api.annotation.download_json_batch(dataset_id, image_ids, progress_cb=p)
# Output:
# {"message": "progress", "event_type": "EventType.PROGRESS", "subtask": "Annotations downloaded: ", "current": 0, "total": 2, "timestamp": "2021-03-16T15:20:06.168Z", "level": "info"}
# {"message": "progress", "event_type": "EventType.PROGRESS", "subtask": "Annotations downloaded: ", "current": 2, "total": 2, "timestamp": "2021-03-16T15:20:06.510Z", "level": "info"}
"""
results = self.download_batch(
dataset_id=dataset_id,
image_ids=image_ids,
progress_cb=progress_cb,
force_metadata_for_links=force_metadata_for_links,
)
return [ann_info.annotation for ann_info in results]
[docs] def upload_path(
self,
img_id: int,
ann_path: str,
skip_bounds_validation: Optional[bool] = False,
) -> None:
"""
Loads an annotation from a given path to a given image ID in the API.
:param img_id: Image ID in Supervisely.
:type img_id: int
:param ann_path: Path to annotation on host.
:type ann_path: str
:return: None
:rtype: :class:`NoneType`
:Usage example:
.. code-block:: python
import supervisely as sly
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
image_id = 121236918
ann_path = '/home/admin/work/supervisely/example/ann.json'
upl_path = api.annotation.upload_path(image_id, ann_path)
"""
self.upload_paths([img_id], [ann_path], skip_bounds_validation=skip_bounds_validation)
[docs] def upload_paths(
self,
img_ids: List[int],
ann_paths: List[str],
progress_cb: Optional[Union[tqdm, Callable]] = None,
skip_bounds_validation: Optional[bool] = False,
) -> None:
"""
Loads an annotations from a given paths to a given images IDs in the API. Images IDs must be from one dataset.
:param img_ids: Images IDs in Supervisely.
:type img_ids: List[int]
:param ann_paths: Paths to annotations on local machine.
:type ann_paths: List[str]
:param progress_cb: Function for tracking download progress.
:type progress_cb: tqdm or callable, optional
:return: None
:rtype: :class:`NoneType`
:Usage example:
.. code-block:: python
import supervisely as sly
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
img_ids = [121236918, 121236919]
ann_pathes = ['/home/admin/work/supervisely/example/ann1.json', '/home/admin/work/supervisely/example/ann2.json']
upl_paths = api.annotation.upload_paths(img_ids, ann_pathes)
# Optimizing the upload process by using the context to avoid redundant API calls.
# Usefull when uploading a large number of annotations in one dataset.
# 1. Download the project meta
dataset_id = 254737
project_id = api.dataset.get_info_by_id(dataset_id).project_id
project_meta = api.project.get_meta(project_id)
# 2. Use the context to avoid redundant API calls
with sly.ApiContext(api, dataset_id=dataset_id, project_id=project_id, project_meta=project_meta):
api.annotation.upload_paths(img_ids, ann_pathes)
"""
def read_json(ann_path):
with open(ann_path) as json_file:
return json.load(json_file)
self._upload_batch(
read_json,
img_ids,
ann_paths,
progress_cb,
skip_bounds_validation=skip_bounds_validation,
)
[docs] def upload_json(
self,
img_id: int,
ann_json: Dict,
skip_bounds_validation: Optional[bool] = False,
) -> None:
"""
Loads an annotation from dict to a given image ID in the API.
:param img_id: Image ID in Supervisely.
:type img_id: int
:param ann_json: Annotation in JSON format.
:type ann_json: dict
:return: None
:rtype: :class:`NoneType`
:Usage example:
.. code-block:: python
import supervisely as sly
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
image_id = 121236918
upl_json = api.annotation.upload_json(image_id, ann_json)
"""
self.upload_jsons([img_id], [ann_json], skip_bounds_validation=skip_bounds_validation)
[docs] def upload_jsons(
self,
img_ids: List[int],
ann_jsons: List[Dict],
progress_cb: Optional[Union[tqdm, Callable]] = None,
skip_bounds_validation: Optional[bool] = False,
) -> None:
"""
Loads an annotations from dicts to a given images IDs in the API. Images IDs must be from one dataset.
:param img_ids: Image ID in Supervisely.
:type img_ids: List[int]
:param ann_jsons: Annotation in JSON format.
:type ann_jsons: List[dict]
:param progress_cb: Function for tracking download progress.
:type progress_cb: tqdm or callable, optional
:return: None
:rtype: :class:`NoneType`
:Usage example:
.. code-block:: python
import supervisely as sly
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
img_ids = [121236918, 121236919]
api.annotation.upload_jsons(img_ids, ann_jsons)
# Optimizing the upload process by using the context to avoid redundant API calls.
# Usefull when uploading a large number of annotations in one dataset.
# 1. Download the project meta
dataset_id = 254737
project_id = api.dataset.get_info_by_id(dataset_id).project_id
project_meta = api.project.get_meta(project_id)
# 2. Use the context to avoid redundant API calls
with sly.ApiContext(api, dataset_id=dataset_id, project_id=project_id, project_meta=project_meta):
api.annotation.upload_jsons(img_ids, ann_jsons)
"""
self._upload_batch(
lambda x: x,
img_ids,
ann_jsons,
progress_cb,
skip_bounds_validation=skip_bounds_validation,
)
[docs] def upload_ann(
self,
img_id: int,
ann: Annotation,
skip_bounds_validation: Optional[bool] = False,
) -> None:
"""
Loads an :class:`Annotation<supervisely.annotation.annotation.Annotation>` to a given image ID in the API.
:param img_id: Image ID in Supervisely.
:type img_id: int
:param ann: Annotation object.
:type ann: Annotation
:return: None
:rtype: :class:`NoneType`
:Usage example:
.. code-block:: python
import supervisely as sly
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
image_id = 121236918
upl_ann = api.annotation.upload_ann(image_id, ann)
"""
self.upload_anns([img_id], [ann], skip_bounds_validation=skip_bounds_validation)
[docs] def upload_anns(
self,
img_ids: List[int],
anns: List[Annotation],
progress_cb: Optional[Union[tqdm, Callable]] = None,
skip_bounds_validation: Optional[bool] = False,
) -> None:
"""
Loads an :class:`Annotations<supervisely.annotation.annotation.Annotation>` to a given images IDs in the API. Images IDs must be from one dataset.
:param img_ids: Image ID in Supervisely.
:type img_ids: List[int]
:param anns: List of Annotation objects.
:type anns: List[Annotation]
:param progress_cb: Function for tracking download progress.
:type progress_cb: tqdm or callable, optional
:return: None
:rtype: :class:`NoneType`
:Usage example:
.. code-block:: python
import supervisely as sly
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
img_ids = [121236918, 121236919]
upl_anns = api.annotation.upload_anns(img_ids, [ann1, ann2])
# Optimizing the upload process by using the context to avoid redundant API calls.
# Usefull when uploading a large number of annotations in one dataset.
# 1. Download the project meta
dataset_id = 254737
project_id = api.dataset.get_info_by_id(dataset_id).project_id
project_meta = api.project.get_meta(project_id)
# 2. Use the context to avoid redundant API calls
with sly.ApiContext(api, dataset_id=dataset_id, project_id=project_id, project_meta=project_meta):
api.annotation.upload_anns(img_ids, [ann1, ann2])
"""
# img_ids from the same dataset
self._upload_batch(
Annotation.to_json,
img_ids,
anns,
progress_cb,
skip_bounds_validation=skip_bounds_validation,
)
def _upload_batch(
self,
func_ann_to_json: Callable,
img_ids: List[int],
anns: List[Union[Dict, Annotation, str]],
progress_cb=None,
skip_bounds_validation: Optional[bool] = False,
):
"""
General method for uploading annotations to instance.
Method is used in: upload_paths, upload_jsons, upload_anns
:param func_ann_to_json: Function to convert annotation to json or read annotation from file.
:type func_ann_to_json: callable
:param img_ids: List of image IDs in Supervisely to which annotations will be uploaded.
:type img_ids: List[int]
:param anns: List of annotations. Can be json, Annotation object or path to annotation file.
:type anns: List[Union[Dict, Annotation, str]]
:param progress_cb: Function for tracking download progress.
:type progress_cb: tqdm or callable, optional
:param skip_bounds_validation: Skip bounds validation.
:type skip_bounds_validation: bool, optional
:return: None
:rtype: :class:`NoneType`
"""
# img_ids from the same dataset
if len(img_ids) == 0:
return
if len(img_ids) != len(anns):
raise RuntimeError(
'Can not match "img_ids" and "anns" lists, len(img_ids) != len(anns)'
)
# use context to avoid redundant API calls
dataset_id = self._api.image.get_info_by_id(
img_ids[0], force_metadata_for_links=False
).dataset_id
context = self._api.optimization_context
context_dataset_id = context.get("dataset_id")
project_id = context.get("project_id")
project_meta = context.get("project_meta")
if dataset_id != context_dataset_id:
context["dataset_id"] = dataset_id
project_id, project_meta = None, None
if not isinstance(project_meta, ProjectMeta):
if project_id is None:
project_id = self._api.dataset.get_info_by_id(dataset_id).project_id
context["project_id"] = project_id
project_meta = ProjectMeta.from_json(self._api.project.get_meta(project_id))
context["project_meta"] = project_meta
need_upload_alpha_masks = False
for obj_cls in project_meta.obj_classes:
if obj_cls.geometry_type == AlphaMask:
need_upload_alpha_masks = True
break
for batch in batched(list(zip(img_ids, anns))):
data = []
if need_upload_alpha_masks:
special_figures = []
special_geometries = []
# check if there are any AlphaMask geometries in the batch
for img_id, ann in batch:
ann_json = func_ann_to_json(ann)
ann_json = deepcopy(ann_json) # Avoid changing the original data
ann_json = Annotation._to_subpixel_coordinate_system_json(ann_json)
filtered_labels = []
if AnnotationJsonFields.LABELS not in ann_json:
raise RuntimeError(
f"Annotation JSON does not contain '{AnnotationJsonFields.LABELS}' field"
)
for label_json in ann_json[AnnotationJsonFields.LABELS]:
for key in [
LabelJsonFields.GEOMETRY_TYPE,
LabelJsonFields.OBJ_CLASS_NAME,
]:
if key not in label_json:
raise RuntimeError(f"Label JSON does not contain '{key}' field")
if label_json[LabelJsonFields.GEOMETRY_TYPE] == AlphaMask.geometry_name():
label_json.update({ApiField.ENTITY_ID: img_id})
obj_cls_name = label_json.get(LabelJsonFields.OBJ_CLASS_NAME)
obj_cls = project_meta.get_obj_class(obj_cls_name)
if obj_cls is None:
raise RuntimeError(
f"Object class '{obj_cls_name}' not found in project meta"
)
# update obj class id in label json
label_json[LabelJsonFields.OBJ_CLASS_ID] = obj_cls.sly_id
geometry = label_json.pop(
BITMAP
) # remove alpha mask geometry from label json
special_geometries.append(geometry)
special_figures.append(label_json)
else:
filtered_labels.append(label_json)
if len(filtered_labels) != len(ann_json[AnnotationJsonFields.LABELS]):
ann_json[AnnotationJsonFields.LABELS] = filtered_labels
data.append({ApiField.IMAGE_ID: img_id, ApiField.ANNOTATION: ann_json})
else:
for img_id, ann in batch:
ann_json = func_ann_to_json(ann)
ann_json = deepcopy(ann_json) # Avoid changing the original data
ann_json = Annotation._to_subpixel_coordinate_system_json(ann_json)
data.append({ApiField.IMAGE_ID: img_id, ApiField.ANNOTATION: ann_json})
self._api.post(
"annotations.bulk.add",
data={
ApiField.DATASET_ID: dataset_id,
ApiField.ANNOTATIONS: data,
ApiField.SKIP_BOUNDS_VALIDATION: skip_bounds_validation,
},
)
if need_upload_alpha_masks:
if len(special_figures) > 0:
# 1. create figures
json_body = {
ApiField.DATASET_ID: dataset_id,
ApiField.FIGURES: special_figures,
ApiField.SKIP_BOUNDS_VALIDATION: skip_bounds_validation,
}
resp = self._api.post("figures.bulk.add", json_body)
added_fig_ids = [resp_obj[ApiField.ID] for resp_obj in resp.json()]
# 2. upload alpha mask geometries
self._api.image.figure.upload_geometries_batch(
added_fig_ids, special_geometries
)
if progress_cb is not None:
progress_cb(len(batch))
[docs] def get_info_by_id(self, id):
"""
get_info_by_id
"""
raise NotImplementedError("Method is not supported")
[docs] def get_info_by_name(self, parent_id, name):
"""
get_info_by_name
"""
raise NotImplementedError("Method is not supported")
[docs] def exists(self, parent_id, name):
"""
exists
"""
raise NotImplementedError("Method is not supported")
[docs] def get_free_name(self, parent_id, name):
"""
get_free_name
"""
raise NotImplementedError("Method is not supported")
def _add_sort_param(self, data):
"""
_add_sort_param
"""
return data
[docs] def copy_batch(
self,
src_image_ids: List[int],
dst_image_ids: List[int],
progress_cb: Optional[Union[tqdm, Callable]] = None,
force_metadata_for_links: Optional[bool] = True,
skip_bounds_validation: Optional[bool] = False,
) -> None:
"""
Copy annotations from one images IDs to another in API.
:param src_image_ids: Images IDs in Supervisely.
:type src_image_ids: List[int]
:param dst_image_ids: Unique IDs of images in API.
:type dst_image_ids: List[int]
:param progress_cb: Function for tracking download progress.
:type progress_cb: tqdm or callable, optional
:raises: :class:`RuntimeError`, if len(src_image_ids) != len(dst_image_ids)
:return: None
:rtype: :class:`NoneType`
:Usage example:
.. code-block:: python
import supervisely as sly
from tqdm import tqdm
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
src_ids = [121236918, 121236919]
dst_ids = [547837053, 547837054]
p = tqdm(desc="Annotations copy: ", total=len(src_ids))
copy_anns = api.annotation.copy_batch(src_ids, dst_ids, progress_cb=p)
# Output:
# {"message": "progress", "event_type": "EventType.PROGRESS", "subtask": "Annotations copy: ", "current": 0, "total": 2, "timestamp": "2021-03-16T15:24:31.286Z", "level": "info"}
# {"message": "progress", "event_type": "EventType.PROGRESS", "subtask": "Annotations copy: ", "current": 2, "total": 2, "timestamp": "2021-03-16T15:24:31.288Z", "level": "info"}
"""
if len(src_image_ids) != len(dst_image_ids):
raise RuntimeError(
'Can not match "src_image_ids" and "dst_image_ids" lists, '
"len(src_image_ids) != len(dst_image_ids)"
)
if len(src_image_ids) == 0:
return
src_dataset_id = self._api.image.get_info_by_id(src_image_ids[0]).dataset_id
for cur_batch in batched(list(zip(src_image_ids, dst_image_ids))):
src_ids_batch, dst_ids_batch = zip(*cur_batch)
ann_infos = self.download_batch(
src_dataset_id,
src_ids_batch,
force_metadata_for_links=force_metadata_for_links,
)
ann_jsons = [ann_info.annotation for ann_info in ann_infos]
self.upload_jsons(
dst_ids_batch, ann_jsons, skip_bounds_validation=skip_bounds_validation
)
if progress_cb is not None:
progress_cb(len(src_ids_batch))
[docs] def copy(
self,
src_image_id: int,
dst_image_id: int,
force_metadata_for_links: Optional[bool] = True,
skip_bounds_validation: Optional[bool] = False,
) -> None:
"""
Copy annotation from one image ID to another image ID in API.
:param src_image_id: Image ID in Supervisely.
:type src_image_id: int
:param dst_image_id: Image ID in Supervisely.
:type dst_image_id: int
:return: None
:rtype: :class:`NoneType`
:Usage example:
.. code-block:: python
import supervisely as sly
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
src_id = 121236918
dst_id = 547837053
api.annotation.copy(src_id, dst_id)
"""
self.copy_batch(
[src_image_id],
[dst_image_id],
force_metadata_for_links=force_metadata_for_links,
skip_bounds_validation=skip_bounds_validation,
)
[docs] def copy_batch_by_ids(
self,
src_image_ids: List[int],
dst_image_ids: List[int],
batch_size: Optional[int] = 50,
save_source_date: Optional[bool] = True,
) -> None:
"""
Copy annotations from one images IDs to another images IDs in API.
:param src_image_ids: Images IDs in Supervisely.
:type src_image_ids: List[int]
:param dst_image_ids: Images IDs in Supervisely.
:type dst_image_ids: List[int]
:return: None
:rtype: :class:`NoneType`
:raises: :class:`RuntimeError` if len(src_image_ids) != len(dst_image_ids)
:Usage example:
.. code-block:: python
import supervisely as sly
os.environ['SERVER_ADDRESS'] = 'https://app.supervisely.com'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()
src_ids = [121236918, 121236919]
dst_ids = [547837053, 547837054]
api.annotation.copy_batch_by_ids(src_ids, dst_ids)
"""
if len(src_image_ids) != len(dst_image_ids):
raise RuntimeError(
'Can not match "src_image_ids" and "dst_image_ids" lists, '
"len(src_image_ids) != len(dst_image_ids)"
)
if len(src_image_ids) == 0:
return
for cur_batch in batched(list(zip(src_image_ids, dst_image_ids)), batch_size=batch_size):
src_ids_batch, dst_ids_batch = zip(*cur_batch)
self._api.post(
"annotations.bulk.copy",
data={
"srcImageIds": src_ids_batch,
"destImageIds": dst_ids_batch,
"preserveSourceDate": save_source_date,
},
)
def _convert_json_info(self, info: dict, skip_missing=True) -> AnnotationInfo:
"""
_convert_json_info
"""
res = super()._convert_json_info(info, skip_missing=skip_missing)
return AnnotationInfo(**res._asdict())
[docs] def append_labels(
self,
image_id: int,
labels: List[Label],
skip_bounds_validation: Optional[bool] = False,
) -> None:
"""
Append labels to image with given ID in API.
:param image_id: Image ID to append labels.
:type image_id: int
:param labels: List of labels to append.
:type labels: List[Label]
:return: None
:rtype: :class:`NoneType`
"""
if len(labels) == 0:
return
alpha_mask_geometry = AlphaMask._impl_json_class_name()
payload = []
special_geometries = {}
for idx, label in enumerate(labels):
_label_json = label.to_json()
if isinstance(label.geometry, AlphaMask):
_label_json.pop(alpha_mask_geometry) # remove alpha mask geometry from label json
special_geometries[idx] = label.geometry.to_json()[BITMAP]
else:
_label_json["geometry"] = label.geometry.to_json()
if "classId" not in _label_json:
raise KeyError("Update project meta from server to get class id")
payload.append(_label_json)
added_ids = []
for batch_jsons in batched(payload, batch_size=100):
resp = self._api.post(
"figures.bulk.add",
{
ApiField.ENTITY_ID: image_id,
ApiField.FIGURES: batch_jsons,
ApiField.SKIP_BOUNDS_VALIDATION: skip_bounds_validation,
},
)
for resp_obj in resp.json():
figure_id = resp_obj[ApiField.ID]
added_ids.append(figure_id)
# upload alpha mask geometries
if len(special_geometries) > 0:
fidure_ids = [added_ids[idx] for idx in special_geometries.keys()]
self._api.image.figure.upload_geometries_batch(fidure_ids, special_geometries.values())
[docs] def get_label_by_id(
self, label_id: int, project_meta: ProjectMeta, with_tags: Optional[bool] = True
) -> Label:
"""Returns Supervisely Label object by it's ID.
:param label_id: ID of the label to get
:type label_id: int
:param project_meta: Supervisely ProjectMeta object
:type project_meta: ProjectMeta
:param with_tags: If True, tags will be added to the Label object
:type with_tags: bool, optional
:return: Supervisely Label object
:rtype: Label
:Usage example:
.. code-block:: python
import os
from dotenv import load_dotenv
import supervisely as sly
# Load secrets and create API object from .env file (recommended)
# Learn more here: https://developer.supervisely.com/getting-started/basics-of-authentication
load_dotenv(os.path.expanduser("~/supervisely.env"))
api = sly.Api.from_env()
label_id = 121236918
project_id = 254737
project_meta = sly.ProjectMeta.from_json(api.project.get_meta(project_id))
label = api.annotation.get_label_by_id(label_id, project_meta)
"""
resp = self._api.get(
"figures.info", {ApiField.ID: label_id, ApiField.DECOMPRESS_BITMAP: False}
)
geometry = resp.json()
class_id = geometry.get("classId")
geometry["classTitle"] = project_meta.get_obj_class_by_id(class_id).name
geometry.update(geometry.get("geometry"))
geometry["tags"] = self._get_label_tags(label_id) if with_tags else []
return Label.from_json(geometry, project_meta)
def _get_label_tags(self, label_id: int) -> List[Dict[str, Any]]:
"""Returns tags of the label with given ID in JSON format.
:param label_id: ID of the label to get tags
:type label_id: int
:return: list of tags in JSON format
:rtype: List[Dict[str, Any]]
:Usage example:
.. code-block:: python
import os
from dotenv import load_dotenv
import supervisely as sly
# Load secrets and create API object from .env file (recommended)
# Learn more here: https://developer.supervisely.com/getting-started/basics-of-authentication
load_dotenv(os.path.expanduser("~/supervisely.env"))
api = sly.Api.from_env()
label_id = 121236918
tags_json = api.annotation.get_label_tags(label_id)
"""
return self._api.get("figures.tags.list", {ApiField.ID: label_id}).json()
[docs] def update_label(self, label_id: int, label: Label) -> None:
"""Updates label with given ID in Supervisely with new Label object.
NOTE: This method only updates label's geometry and tags, not class title, etc.
:param label_id: ID of the label to update
:type label_id: int
:param label: Supervisely Label object
:type label: Label
:Usage example:
.. code-block:: python
import os
from dotenv import load_dotenv
import supervisely as sly
# Load secrets and create API object from .env file (recommended)
# Learn more here: https://developer.supervisely.com/getting-started/basics-of-authentication
load_dotenv(os.path.expanduser("~/supervisely.env"))
api = sly.Api.from_env()
new_label: sly.Label
label_id = 121236918
api.annotation.update_label(label_id, new_label)
"""
self._api.post(
"figures.editInfo",
{
ApiField.ID: label_id,
ApiField.TAGS: [tag.to_json() for tag in label.tags],
ApiField.GEOMETRY: label.geometry.to_json(),
},
)
[docs] def update_label_priority(self, label_id: int, priority: int) -> None:
"""Updates label's priority with given ID in Supervisely.
Priority increases with the number: a higher number indicates a higher priority.
The higher priority means that the label will be displayed on top of the others.
The lower priority means that the label will be displayed below the others.
:param label_id: ID of the label to update
:type label_id: int
:param priority: New priority of the label
:type priority: int
:Usage example:
.. code-block:: python
import os
from dotenv import load_dotenv
import supervisely as sly
# Load secrets and create API object from .env file (recommended)
# Learn more here: https://developer.supervisely.com/getting-started/basics-of-authentication
load_dotenv(os.path.expanduser("~/supervisely.env"))
api = sly.Api.from_env()
label_ids = [123, 456, 789]
priorities = [1, 2, 3]
for label_id, priority in zip(label_ids, priorities):
api.annotation.update_label_priority(label_id, priority)
# The label with ID 789 will be displayed on top of the others.
# The label with ID 123 will be displayed below the others.
"""
self._api.post(
"figures.priority.update",
{
ApiField.ID: label_id,
ApiField.PRIORITY: priority,
},
)