Source code for supervisely.project.video_project

# coding: utf-8

# docs
from __future__ import annotations

import os
from collections import namedtuple
from typing import Callable, Dict, List, NamedTuple, Optional, Tuple, Union

from tqdm import tqdm

from supervisely._utils import batched
from supervisely.api.api import Api
from supervisely.api.module_api import ApiField
from supervisely.api.video.video_api import VideoInfo
from supervisely.collection.key_indexed_collection import KeyIndexedCollection
from supervisely.io.fs import file_exists, mkdir, touch
from supervisely.io.json import dump_json_file, load_json_file
from supervisely.project.project import Dataset, OpenMode, Project
from supervisely.project.project import read_single_project as read_project_wrapper
from supervisely.project.project_meta import ProjectMeta
from supervisely.project.project_type import ProjectType
from supervisely.sly_logger import logger
from supervisely.task.progress import Progress, tqdm_sly
from supervisely.video import video as sly_video
from supervisely.video_annotation.key_id_map import KeyIdMap
from supervisely.video_annotation.video_annotation import VideoAnnotation


[docs]class VideoItemPaths(NamedTuple): #: :class:`str`: Full video file path of item video_path: str #: :class:`str`: Full annotation file path of item ann_path: str
[docs]class VideoDataset(Dataset): """ VideoDataset is where your labeled and unlabeled videos and other data files live. :class:`VideoDataset<VideoDataset>` object is immutable. :param directory: Path to dataset directory. :type directory: str :param mode: Determines working mode for the given dataset. :type mode: :class:`OpenMode<supervisely.project.project.OpenMode>` :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) """ #: :class:`str`: Items data directory name item_dir_name = "video" #: :class:`str`: Annotations directory name ann_dir_name = "ann" #: :class:`str`: Items info directory name item_info_dir_name = "video_info" #: :class:`str`: Segmentation masks directory name seg_dir_name = None annotation_class = VideoAnnotation item_info_class = VideoInfo @property def project_dir(self) -> str: """ Path to the video project containing the video dataset. :return: Path to the video project. :rtype: :class:`str` :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) print(ds.project_dir) # Output: "/home/admin/work/supervisely/projects/videos_example" """ return super().project_dir @property def name(self) -> str: """ Video Dataset name. :return: Video Dataset Name. :rtype: :class:`str` :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) print(ds.name) # Output: "ds0" """ return super().name @property def directory(self) -> str: """ Path to the video dataset directory. :return: Path to the video dataset directory. :rtype: :class:`str` :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) print(ds.directory) # Output: '/home/admin/work/supervisely/projects/videos_example/ds0' """ return super().directory @property def item_dir(self) -> str: """ Path to the video dataset items directory. :return: Path to the video dataset items directory. :rtype: :class:`str` :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) print(ds.item_dir) # Output: '/home/admin/work/supervisely/projects/videos_example/ds0/video' """ return super().item_dir @property def img_dir(self) -> str: """ Not available for VideoDataset class object. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Property 'img_dir' is not supported for {type(self).__name__} object." ) @property def ann_dir(self) -> str: """ Path to the video dataset annotations directory. :return: Path to the video dataset directory with annotations. :rtype: :class:`str` :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) print(ds.ann_dir) # Output: '/home/admin/work/supervisely/projects/videos_example/ds0/ann' """ return super().ann_dir @property def img_info_dir(self): """ Not available for VideoDataset class object. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Property 'img_info_dir' is not supported for {type(self).__name__} object." ) @property def item_info_dir(self): """ Path to the video dataset item with items info. :return: Path to the video dataset directory with items info. :rtype: :class:`str` :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) print(ds.item_info_dir) # Output: '/home/admin/work/supervisely/projects/videos_example/ds0/video_info' """ return super().item_info_dir @property def seg_dir(self): """ Not available for VideoDataset class object. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Property 'seg_dir' is not supported for {type(self).__name__} object." ) @classmethod def _has_valid_ext(cls, path: str) -> bool: """ Checks if file from given path is supported :param path: str :return: bool """ return sly_video.has_valid_ext(path)
[docs] def get_items_names(self) -> list: """ List of video dataset item names. :return: List of item names. :rtype: :class:`list` [ :class:`str` ] :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) print(ds.get_item_names()) # Output: ['video_0002.mp4', 'video_0005.mp4', 'video_0008.mp4', ...] """ return super().get_items_names()
[docs] def item_exists(self, item_name: str) -> bool: """ Checks if given item name belongs to the video dataset. :param item_name: Item name. :type item_name: :class:`str` :return: True if item exist, otherwise False. :rtype: :class:`bool` :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) ds.item_exists("video_0748") # False ds.item_exists("video_0748.mp4") # True """ return super().item_exists(item_name)
[docs] def get_item_path(self, item_name: str) -> str: """ Path to the given item. :param item_name: Item name. :type item_name: :class:`str` :return: Path to the given item. :rtype: :class:`str` :raises: :class:`RuntimeError` if item not found in the project :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) print(ds.get_item_path("video_0748")) # Output: RuntimeError: Item video_0748 not found in the project. print(ds.get_item_path("video_0748.mp4")) # Output: '/home/admin/work/supervisely/projects/videos_example/ds0/video/video_0748.mp4' """ return super().get_item_path(item_name)
[docs] def get_img_path(self, item_name: str) -> str: """ Not available for VideoDataset class object. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Method 'get_img_path(item_name)' is not supported for {type(self).__name__} object." )
[docs] def get_ann( self, item_name, project_meta: ProjectMeta, key_id_map: Optional[KeyIdMap] = None ) -> VideoAnnotation: """ Read annotation of item from json. :param item_name: Item name. :type item_name: :class:`str` :param project_meta: ProjectMeta object. :type project_meta: :class:`ProjectMeta<supervisely.project.project_meta.ProjectMeta>` :param key_id_map: KeyIdMap object. :type key_id_map: :class:`KeyIdMap<supervisely.video_annotation.key_id_map.KeyIdMap>`, optional :return: VideoAnnotation object. :rtype: :class:`VideoAnnotation<supervisely.video_annotation.video_annotation.VideoAnnotation>` :raises: :class:`RuntimeError` if item not found in the project :Usage example: .. code-block:: python import supervisely as sly project_path = "/home/admin/work/supervisely/projects/videos_example" project = sly.VideoProject(project_path, sly.OpenMode.READ) ds = project.datasets.get('ds0') annotation = ds.get_ann("video_0748", project.meta) # Output: RuntimeError: Item video_0748 not found in the project. annotation = ds.get_ann("video_0748.mp4", project.meta) print(annotation.to_json()) # Output: { # "description": "", # "size": { # "height": 500, # "width": 700 # }, # "key": "e9ef52dbbbbb490aa10f00a50e1fade6", # "tags": [], # "objects": [], # "frames": [{ # "index": 0, # "figures": [] # }] # "framesCount": 1 # } """ ann_path = self.get_ann_path(item_name) return self.annotation_class.load_json_file(ann_path, project_meta, key_id_map)
[docs] def get_ann_path(self, item_name: str) -> str: """ Path to the given annotation json file. :param item_name: Item name. :type item_name: :class:`str` :return: Path to the given annotation json file. :rtype: :class:`str` :raises: :class:`RuntimeError` if item not found in the project :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) print(ds.get_ann_path("video_0748")) # Output: RuntimeError: Item video_0748 not found in the project. print(ds.get_ann_path("video_0748.mp4")) # Output: '/home/admin/work/supervisely/projects/videos_example/ds0/ann/video_0748.mp4.json' """ return super().get_ann_path(item_name)
[docs] def get_img_info_path(self, img_name: str) -> str: """ Not available for VideoDataset class object. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Method 'get_img_info_path(item_name)' is not supported for {type(self).__name__} object." )
[docs] def get_item_info_path(self, item_name: str) -> str: """ Get path to the item info json file without checking if the file exists. :param item_name: Item name. :type item_name: :class:`str` :return: Path to the given item info json file. :rtype: :class:`str` :raises: :class:`RuntimeError` if item not found in the project. :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) print(ds.get_item_info_path("video_0748")) # Output: RuntimeError: Item video_0748 not found in the project. print(ds.get_item_info_path("video_0748.mp4")) # Output: '/home/admin/work/supervisely/projects/videos_example/ds0/video_info/video_0748.mp4.json' """ return super().get_item_info_path(item_name)
[docs] def get_image_info(self, item_name: str) -> None: """ Not available for VideoDataset class object. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Method 'get_image_info(item_name)' is not supported for {type(self).__name__} object." )
[docs] def get_item_info(self, item_name: str) -> VideoInfo: """ Information for Item with given name. :param item_name: Item name. :type item_name: :class:`str` :return: VideoInfo object. :rtype: :class:`VideoInfo<supervisely.api.video.video_api.VideoInfo>` :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) print(ds.get_item_info("video_0748.mp4")) # Output: # VideoInfo( # id=198702499, # name='video_0748.mp4', # hash='ehYHLNFWmMNuF2fPUgnC/g/tkIIEjNIOhdbNLQXkE8Y=', # team_id=16087, # workspace_id=23821, # project_id=124974, # dataset_id=466639, # path_original='/h5un6l2bnaz1vj8a9qgms4-public/videos/w/7/i4/GZYoCs...9F3kyVJ7.mp4', # frames_to_timecodes=[0, 0.033367, 0.066733, 0.1001,...,10.777433, 10.8108, 10.844167], # frames_count=326, # frame_width=3840, # frame_height=2160, # created_at='2021-03-23T13:14:25.536Z', # updated_at='2021-03-23T13:16:43.300Z' # ) """ item_info_path = self.get_item_info_path(item_name) item_info_dict = load_json_file(item_info_path) item_info_named_tuple = namedtuple(self.item_info_class.__name__, item_info_dict) return item_info_named_tuple(**item_info_dict)
[docs] def get_seg_path(self, item_name: str) -> str: """ Not available for VideoDataset class object. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Method 'get_seg_path(item_name)' is not supported for {type(self).__name__} object." )
[docs] def add_item_file( self, item_name: str, item_path: str, ann: Optional[Union[VideoAnnotation, str]] = None, _validate_item: Optional[bool] = True, _use_hardlink: Optional[bool] = False, item_info: Optional[Union[VideoInfo, Dict, str]] = None, ) -> None: """ Adds given item file to dataset items directory, and adds given annotation to dataset annotations directory. if ann is None, creates empty annotation file. :param item_name: Item name. :type item_name: :class:`str` :param item_path: Path to the item. :type item_path: :class:`str` :param ann: VideoAnnotation object or path to annotation json file. :type ann: :class:`VideoAnnotation<supervisely.video_annotation.video_annotation.VideoAnnotation>` or :class:`str`, optional :param _validate_item: Checks input files format. :type _validate_item: :class:`bool`, optional :param _use_hardlink: If True creates a hardlink pointing to src named dst, otherwise don't. :type _use_hardlink: :class:`bool`, optional :param item_info: VideoInfo object or VideoInfo object converted to dict or path to item info json file for copying to dataset item info directory. :type item_info: :class:`VideoInfo<supervisely.api.video.video_api.VideoInfo>` or :class:`dict` or :class:`str`, optional :return: None :rtype: NoneType :raises: :class:`RuntimeError` if item_name already exists in dataset or item name has unsupported extension. :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) ann = "/home/admin/work/supervisely/projects/videos_example/ds0/ann/video_8888.mp4.json" ds.add_item_file("video_8888.mp4", "/home/admin/work/supervisely/projects/videos_example/ds0/video/video_8888.mp4", ann=ann) print(ds.item_exists("video_8888.mp4")) # Output: True """ return super().add_item_file( item_name=item_name, item_path=item_path, ann=ann, _validate_item=_validate_item, _use_hardlink=_use_hardlink, item_info=item_info, )
[docs] def add_item_np(self, item_name, img, ann=None, img_info=None): """ Not available for VideoDataset class object. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Method 'add_item_np()' is not supported for {type(self).__name__} object." )
[docs] def add_item_raw_bytes(self, item_name, item_raw_bytes, ann=None, img_info=None): """ Not available for VideoDataset class object. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Method 'add_item_raw_bytes()' is not supported for {type(self).__name__} object." )
def get_classes_stats( self, project_meta: Optional[ProjectMeta] = None, return_objects_count: Optional[bool] = True, return_figures_count: Optional[bool] = True, return_items_count: Optional[bool] = True, ): if project_meta is None: project = VideoProject(self.project_dir, OpenMode.READ) project_meta = project.meta class_items = {} class_objects = {} class_figures = {} for obj_class in project_meta.obj_classes: class_items[obj_class.name] = 0 class_objects[obj_class.name] = 0 class_figures[obj_class.name] = 0 for item_name in self: item_ann = self.get_ann(item_name, project_meta) item_class = {} for ann_obj in item_ann.objects: class_objects[ann_obj.obj_class.name] += 1 for video_figure in item_ann.figures: class_figures[video_figure.parent_object.obj_class.name] += 1 item_class[video_figure.parent_object.obj_class.name] = True for obj_class in project_meta.obj_classes: if obj_class.name in item_class.keys(): class_items[obj_class.name] += 1 result = {} if return_items_count: result["items_count"] = class_items if return_objects_count: result["objects_count"] = class_objects if return_figures_count: result["figures_count"] = class_figures return result def _get_empty_annotaion(self, item_name): """ Create empty VideoAnnotation for given video :param item_name: str :return: VideoAnnotation class object """ img_size, frames_count = sly_video.get_image_size_and_frames_count(item_name) return self.annotation_class(img_size, frames_count) def _add_item_raw_bytes(self, item_name, item_raw_bytes): """ Not available for VideoDataset class object. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Method '_add_item_raw_bytes()' is not supported for {type(self).__name__} object." ) def _add_img_np(self, item_name, img): """ Not available for VideoDataset class object. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Method '_add_img_np()' is not supported for {type(self).__name__} object." ) def _validate_added_item_or_die(self, item_path): """ Make sure we actually received a valid video file, clean it up and fail if not so. :param item_path: str """ # Make sure we actually received a valid video file, clean it up and fail if not so. try: sly_video.validate_format(item_path) except Exception as e: os.remove(item_path) raise e
[docs] def set_ann( self, item_name: str, ann: VideoAnnotation, key_id_map: Optional[KeyIdMap] = None ) -> None: """ Replaces given annotation for given item name to dataset annotations directory in json format. :param item_name: Item name. :type item_name: :class:`str` :param ann: VideoAnnotation object. :type ann: :class:`VideoAnnotation<supervisely.video_annotation.video_annotation.VideoAnnotation>` :param key_id_map: KeyIdMap object. :type key_id_map: :class:`KeyIdMap<supervisely.video_annotation.key_id_map.KeyIdMap>`, optional :return: None :rtype: NoneType :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) height, width = 500, 700 new_ann = sly.VideoAnnotation((height, width), frames_count=0) ds.set_ann("video_0748.mp4", new_ann) """ if type(ann) is not self.annotation_class: raise TypeError( f"Type of 'ann' should be {self.annotation_class.__name__}, not a {type(ann).__name__}" ) dst_ann_path = self.get_ann_path(item_name) dump_json_file(ann.to_json(key_id_map), dst_ann_path, indent=4)
[docs] def get_item_paths(self, item_name) -> VideoItemPaths: """ Generates :class:`VideoItemPaths<VideoItemPaths>` object with paths to item and annotation directories for item with given name. :param item_name: Item name. :type item_name: :class:`str` :return: VideoItemPaths object :rtype: :class:`VideoItemPaths<VideoItemPaths>` :Usage example: .. code-block:: python import supervisely as sly dataset_path = "/home/admin/work/supervisely/projects/videos_example/ds0" ds = sly.VideoDataset(dataset_path, sly.OpenMode.READ) video_path, ann_path = dataset.get_item_paths("video_0748.mp4") print("video_path:", video_path) print("ann_path:", ann_path) # Output: # video_path: /home/admin/work/supervisely/projects/videos_example/ds0/video/video_0748.mp4 # ann_path: /home/admin/work/supervisely/projects/videos_example/ds0/ann/video_0748.mp4.json """ return VideoItemPaths( video_path=self.get_item_path(item_name), ann_path=self.get_ann_path(item_name) )
[docs] @staticmethod def get_url(project_id: int, dataset_id: int) -> str: """ Get URL to dataset items list in Supervisely. :param project_id: :class:`VideoProject<VideoProject>` ID in Supervisely. :type project_id: :class:`int` :param dataset_id: :class:`VideoDataset<VideoDataset>` ID in Supervisely. :type dataset_id: :class:`int` :return: URL to dataset items list. :rtype: :class:`str` :Usage example: .. code-block:: python from supervisely import VideoDataset project_id = 10093 dataset_id = 45330 ds_items_link = VideoDataset.get_url(project_id, dataset_id) print(ds_items_link) # Output: "/projects/10093/datasets/45330/entities" """ return super().get_url(project_id, dataset_id)
[docs]class VideoProject(Project): """ VideoProject is a parent directory for video dataset. VideoProject object is immutable. :param directory: Path to video project directory. :type directory: :class:`str` :param mode: Determines working mode for the given project. :type mode: :class:`OpenMode<supervisely.project.project.OpenMode>` :Usage example: .. code-block:: python import supervisely as sly project_path = "/home/admin/work/supervisely/projects/videos_example" project = sly.Project(project_path, sly.OpenMode.READ) """ dataset_class = VideoDataset class DatasetDict(KeyIndexedCollection): item_type = VideoDataset def __init__(self, directory, mode: OpenMode): """ :param directory: path to the directory where the project will be saved or where it will be loaded from :param mode: OpenMode class object which determines in what mode to work with the project (generate exception error if not so) """ self._key_id_map: KeyIdMap = None super().__init__(directory, mode)
[docs] @staticmethod def get_url(id: int) -> str: """ Get URL to video datasets list in Supervisely. :param id: :class:`VideoProject<VideoProject>` ID in Supervisely. :type id: :class:`int` :return: URL to datasets list. :rtype: :class:`str` :Usage example: .. code-block:: python from supervisely import VideoProject project_id = 10093 datasets_link = VideoProject.get_url(project_id) print(datasets_link) # Output: "/projects/10093/datasets" """ return super().get_url(id)
def get_classes_stats( self, dataset_names: Optional[List[str]] = None, return_objects_count: Optional[bool] = True, return_figures_count: Optional[bool] = True, return_items_count: Optional[bool] = True, ): return super(VideoProject, self).get_classes_stats( dataset_names, return_objects_count, return_figures_count, return_items_count ) def _read(self): """ Download project from given project directory. Checks item and annotation directoris existing and dataset not empty. Consistency checks. Every video must have an annotation, and the correspondence must be one to one. """ super()._read() self._key_id_map = KeyIdMap() if os.path.exists(self._get_key_id_map_path()): self._key_id_map = self._key_id_map.load_json(self._get_key_id_map_path()) def _create(self): """ Creates a leaf directory and empty meta.json file. Generate exception error if project directory already exists and is not empty. """ super()._create() self.set_key_id_map(KeyIdMap()) @property def key_id_map(self): # TODO: write docstring return self._key_id_map @property def type(self) -> str: """ Project type. :return: Project type. :rtype: :class:`str` :Usage example: .. code-block:: python import supervisely as sly project = sly.VideoProject("/home/admin/work/supervisely/projects/video", sly.OpenMode.READ) print(project.type) # Output: 'videos' """ return ProjectType.VIDEOS.value
[docs] def set_key_id_map(self, new_map: KeyIdMap): """ Save given KeyIdMap object to project dir in json format. :param new_map: KeyIdMap class object """ self._key_id_map = new_map self._key_id_map.dump_json(self._get_key_id_map_path())
def _get_key_id_map_path(self): """ :return: str (full path to key_id_map.json) """ return os.path.join(self.directory, "key_id_map.json")
[docs] def copy_data( self, dst_directory: str, dst_name: Optional[str] = None, _validate_item: Optional[bool] = True, _use_hardlink: Optional[bool] = False, ) -> VideoProject: """ Makes a copy of the :class:`VideoProject<VideoProject>`. :param dst_directory: Path to video project parent directory. :type dst_directory: :class:`str` :param dst_name: Video Project name. :type dst_name: :class:`str`, optional :param _validate_item: Checks input files format. :type _validate_item: :class:`bool`, optional :param _use_hardlink: If True creates a hardlink pointing to src named dst, otherwise don't. :type _use_hardlink: :class:`bool`, optional :return: VideoProject object. :rtype: :class:`VideoProject<VideoProject>` :Usage example: .. code-block:: python import supervisely as sly project = sly.VideoProject("/home/admin/work/supervisely/projects/videos_example", sly.OpenMode.READ) print(project.total_items) # Output: 6 new_project = project.copy_data("/home/admin/work/supervisely/projects/", "videos_example_copy") print(new_project.total_items) # Output: 6 """ dst_name = dst_name if dst_name is not None else self.name new_project = VideoProject(os.path.join(dst_directory, dst_name), OpenMode.CREATE) new_project.set_meta(self.meta) for ds in self: new_ds = new_project.create_dataset(ds.name) for item_name in ds: item_path, ann_path = ds.get_item_paths(item_name) item_info_path = ds.get_item_info_path(item_name) item_path = item_path if os.path.isfile(item_path) else None ann_path = ann_path if os.path.isfile(ann_path) else None item_info_path = item_info_path if os.path.isfile(item_info_path) else None try: new_ds.add_item_file( item_name, item_path, ann_path, _validate_item=_validate_item, _use_hardlink=_use_hardlink, item_info=item_info_path, ) except Exception as e: logger.info( "INFO FOR DEBUGGING", extra={ "source_project_name": self.name, "dst_directory": dst_directory, "ds_name": ds.name, "item_name": item_name, "item_path": item_path, "ann_path": ann_path, "item_info": item_info_path, }, ) raise e new_project.set_key_id_map(self.key_id_map) return new_project
[docs] @staticmethod def to_segmentation_task( src_project_dir: str, dst_project_dir: Optional[str] = None, inplace: Optional[bool] = False, target_classes: Optional[List[str]] = None, progress_cb: Optional[Union[tqdm, Callable]] = None, segmentation_type: Optional[str] = "semantic", ) -> None: """ Not available for VideoProject class. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Static method 'to_segmentation_task()' is not supported for VideoProject class now." )
[docs] @staticmethod def to_detection_task( src_project_dir: str, dst_project_dir: Optional[str] = None, inplace: Optional[bool] = False, progress_cb: Optional[Union[tqdm, Callable]] = None, ) -> None: """ Not available for VideoProject class. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Static method 'to_detection_task()' is not supported for VideoProject class now." )
[docs] @staticmethod def remove_classes_except( project_dir: str, classes_to_keep: Optional[List[str]] = None, inplace: Optional[bool] = False, ) -> None: """ Not available for VideoProject class. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Static method 'remove_classes_except()' is not supported for VideoProject class now." )
[docs] @staticmethod def remove_classes( project_dir: str, classes_to_remove: Optional[List[str]] = None, inplace: Optional[bool] = False, ) -> None: """ Not available for VideoProject class. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Static method 'remove_classes()' is not supported for VideoProject class now." )
@staticmethod def _remove_items( project_dir, without_objects=False, without_tags=False, without_objects_and_tags=False, inplace=False, ): """ Not available for VideoProject class. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Static method '_remove_items()' is not supported for VideoProject class now." )
[docs] @staticmethod def remove_items_without_objects(project_dir: str, inplace: Optional[bool] = False) -> None: """ Not available for VideoProject class. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Static method 'remove_items_without_objects()' is not supported for VideoProject class now." )
[docs] @staticmethod def remove_items_without_tags(project_dir: str, inplace: Optional[bool] = False) -> None: """ Not available for VideoProject class. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Static method 'remove_items_without_tags()' is not supported for VideoProject class now." )
[docs] @staticmethod def remove_items_without_both_objects_and_tags( project_dir: str, inplace: Optional[bool] = False ) -> None: """ Not available for VideoProject class. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Static method 'remove_items_without_both_objects_and_tags()' is not supported for VideoProject class now." )
[docs] @staticmethod def get_train_val_splits_by_count(project_dir: str, train_count: int, val_count: int) -> None: """ Not available for VideoProject class. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Static method 'get_train_val_splits_by_count()' is not supported for VideoProject class now." )
[docs] @staticmethod def get_train_val_splits_by_tag( project_dir: str, train_tag_name: str, val_tag_name: str, untagged: Optional[str] = "ignore", ) -> None: """ Not available for VideoProject class. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Static method 'get_train_val_splits_by_tag()' is not supported for VideoProject class now." )
[docs] @staticmethod def get_train_val_splits_by_dataset( project_dir: str, train_datasets: List[str], val_datasets: List[str] ) -> None: """ Not available for VideoProject class. :raises: :class:`NotImplementedError` in all cases. """ raise NotImplementedError( f"Static method 'get_train_val_splits_by_tag()' is not supported for VideoProject class now." )
[docs] @classmethod def read_single(cls, dir): """ Read project from given ditectory. Generate exception error if given dir contains more than one subdirectory :param dir: str :return: VideoProject class object """ return read_project_wrapper(dir, cls)
[docs] @staticmethod def download( api: Api, project_id: int, dest_dir: str, dataset_ids: List[int] = None, download_videos: bool = True, save_video_info: bool = False, log_progress: bool = True, progress_cb: Optional[Union[tqdm, Callable]] = None, ) -> None: """ Download video project from Supervisely to the given directory. :param api: Supervisely Api class object. :type api: :class:`Api<supervisely.api.api.Api>` :param project_id: Project ID in Supervisely. :type project_id: :class:`int` :param dest_dir: Directory to download video project. :type dest_dir: :class:`str` :param dataset_ids: Datasets IDs in Supervisely to download. :type dataset_ids: :class:`list` [ :class:`int` ], optional :param download_videos: Download videos from Supervisely video project in dest_dir or not. :type download_videos: :class:`bool`, optional :param save_video_info: Save video infos or not. :type save_video_info: :class:`bool`, optional :param log_progress: Log download progress or not. :type log_progress: :class:`bool` :param progress_cb: Function for tracking download progress. :type progress_cb: :class:`tqdm`, optional :return: None :rtype: NoneType :Usage example: .. code-block:: python import supervisely as sly # Local destination Project folder save_directory = "/home/admin/work/supervisely/source/video_project" # Obtain server address and your api_token from environment variables # Edit those values if you run this notebook on your own PC address = os.environ['SERVER_ADDRESS'] token = os.environ['API_TOKEN'] # Initialize API object api = sly.Api(address, token) project_id = 8888 # Download Video Project sly.VideoProject.download(api, project_id, save_directory) project_fs = sly.VideoProject(save_directory, sly.OpenMode.READ) """ download_video_project( api=api, project_id=project_id, dest_dir=dest_dir, dataset_ids=dataset_ids, download_videos=download_videos, save_video_info=save_video_info, log_progress=log_progress, progress_cb=progress_cb, )
[docs] @staticmethod def upload( dir: str, api: Api, workspace_id: int, project_name: Optional[str] = None, log_progress: bool = True, progress_cb: Optional[Union[tqdm, Callable]] = None, ) -> Tuple[int, str]: """ Upload video project from given directory in Supervisely. :param dir: Directory with video project. :type dir: str :param api: Api class object. :type api: Api :param workspace_id: Workspace ID in Supervisely to upload video project. :type workspace_id: int :param project_name: Name of video project. :type project_name: str :param log_progress: Logging progress of download video project or not. :type log_progress: bool :return: New video project ID in Supervisely and project name :rtype: :class:`int`, :class:`str` :Usage example: .. code-block:: python import supervisely as sly # Local folder with Video Project project_directory = "/home/admin/work/supervisely/source/video_project" # Obtain server address and your api_token from environment variables # Edit those values if you run this notebook on your own PC address = os.environ['SERVER_ADDRESS'] token = os.environ['API_TOKEN'] # Initialize API object api = sly.Api(address, token) # Upload Video Project project_id, project_name = sly.VideoProject.upload( project_directory, api, workspace_id=45, project_name="My Video Project" ) """ return upload_video_project( dir=dir, api=api, workspace_id=workspace_id, project_name=project_name, log_progress=log_progress, progress_cb=progress_cb, )
def download_video_project( api: Api, project_id: int, dest_dir: str, dataset_ids: Optional[List[int]] = None, download_videos: Optional[bool] = True, save_video_info: Optional[bool] = False, log_progress: bool = True, progress_cb: Optional[Union[tqdm, Callable]] = None, include_custom_data: Optional[bool] = False, ) -> None: """ Download video project to the local directory. :param api: Supervisely API address and token. :type api: Api :param project_id: Project ID to download :type project_id: int :param dest_dir: Destination path to local directory. :type dest_dir: str :param dataset_ids: Specified list of Dataset IDs which will be downloaded. Datasets could be downloaded from different projects but with the same data type. :type dataset_ids: list(int), optional :param download_videos: Include videos in the download. :type download_videos: bool, optional :param save_video_info: Include video info in the download. :type save_video_info: bool, optional :param log_progress: Show downloading logs in the output. :type log_progress: bool :param progress_cb: Function for tracking the download progress. :type progress_cb: tqdm or callable, optional :return: None. :rtype: NoneType :Usage example: .. code-block:: python import os from dotenv import load_dotenv from tqdm import tqdm 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") dest_dir = 'your/local/dest/dir' # Download video project project_id = 17758 project_info = api.project.get_info_by_id(project_id) num_videos = project_info.items_count p = tqdm(desc="Downloading video project", total=num_videos) sly.download( api, project_id, dest_dir, progress_cb=p, ) """ LOG_BATCH_SIZE = 1 key_id_map = KeyIdMap() project_fs = VideoProject(dest_dir, OpenMode.CREATE) meta = ProjectMeta.from_json(api.project.get_meta(project_id)) project_fs.set_meta(meta) if progress_cb is not None: log_progress = False datasets = [] if dataset_ids is not None: for ds_id in dataset_ids: datasets.append(api.dataset.get_info_by_id(ds_id)) else: datasets = api.dataset.get_list(project_id) for dataset in datasets: dataset_fs = project_fs.create_dataset(dataset.name) videos = api.video.get_list(dataset.id) ds_progress = progress_cb if log_progress: ds_progress = tqdm_sly( desc="Downloading videos from {!r}".format(dataset.name), total=len(videos), ) for batch in batched(videos, batch_size=LOG_BATCH_SIZE): video_ids = [video_info.id for video_info in batch] video_names = [video_info.name for video_info in batch] custom_datas = [video_info.custom_data for video_info in batch] try: ann_jsons = api.video.annotation.download_bulk(dataset.id, video_ids) except Exception as e: logger.info( "INFO FOR DEBUGGING", extra={ "project_id": project_id, "dataset_id": dataset.id, "video_ids": video_ids, }, ) raise e for video_id, video_name, custom_data, ann_json, video_info in zip( video_ids, video_names, custom_datas, ann_jsons, batch ): if video_name != ann_json[ApiField.VIDEO_NAME]: raise RuntimeError("Error in api.video.annotation.download_batch: broken order") video_file_path = dataset_fs.generate_item_path(video_name) if include_custom_data: CUSTOM_DATA_DIR = os.path.join(dest_dir, dataset.name, "custom_data") mkdir(CUSTOM_DATA_DIR) custom_data_path = os.path.join(CUSTOM_DATA_DIR, f"{video_name}.json") dump_json_file(custom_data, custom_data_path) if download_videos: try: video_file_size = video_info.file_meta.get("size") if ds_progress is not None and video_file_size is not None: item_progress = tqdm_sly( desc=f"Downloading '{video_name}'", total=int(video_file_size), unit="B", unit_scale=True, leave=False, ) api.video.download_path(video_id, video_file_path, item_progress) else: api.video.download_path(video_id, video_file_path) except Exception as e: logger.info( "INFO FOR DEBUGGING", extra={ "project_id": project_id, "dataset_id": dataset.id, "video_id": video_id, "video_file_path": video_file_path, }, ) raise e else: touch(video_file_path) item_info = video_info._asdict() if save_video_info else None try: video_ann = VideoAnnotation.from_json(ann_json, project_fs.meta, key_id_map) except Exception as e: logger.info( "INFO FOR DEBUGGING", extra={ "project_id": project_id, "dataset_id": dataset.id, "video_id": video_id, "video_name": video_name, "ann_json": ann_json, }, ) raise e try: dataset_fs.add_item_file( video_name, video_file_path, ann=video_ann, _validate_item=False, _use_hardlink=True, item_info=item_info, ) except Exception as e: logger.info( "INFO FOR DEBUGGING", extra={ "project_id": project_id, "dataset_id": dataset.id, "video_id": video_id, "video_name": video_name, "video_file_path": video_file_path, "item_info": item_info, }, ) raise e if progress_cb is not None: progress_cb(1) if log_progress: ds_progress(len(batch)) project_fs.set_key_id_map(key_id_map) def upload_video_project( dir: str, api: Api, workspace_id: int, project_name: Optional[str] = None, log_progress: bool = True, include_custom_data: Optional[bool] = False, progress_cb: Optional[Union[tqdm, Callable]] = None, ) -> Tuple[int, str]: project_fs = VideoProject.read_single(dir) if project_name is None: project_name = project_fs.name if api.project.exists(workspace_id, project_name): project_name = api.project.get_free_name(workspace_id, project_name) project = api.project.create(workspace_id, project_name, ProjectType.VIDEOS) api.project.update_meta(project.id, project_fs.meta.to_json()) if progress_cb is not None: log_progress = False for dataset_fs in project_fs.datasets: dataset_fs: VideoDataset dataset = api.dataset.create(project.id, dataset_fs.name) names, item_paths, ann_paths = [], [], [] for item_name in dataset_fs: video_path, ann_path = dataset_fs.get_item_paths(item_name) names.append(item_name) item_paths.append(video_path) ann_paths.append(ann_path) ds_progress = progress_cb if log_progress is True: ds_progress = tqdm_sly( desc="Uploading videos to {!r}".format(dataset.name), total=len(item_paths), position=0, ) try: item_infos = api.video.upload_paths(dataset.id, names, item_paths, ds_progress) video_ids = [item_info.id for item_info in item_infos] if include_custom_data: for item_info in item_infos: item_name = item_info.name custom_data_path = os.path.join( dir, dataset_fs.name, "custom_data", f"{item_name}.json" ) if os.path.exists(custom_data_path): custom_data = load_json_file(custom_data_path) api.video.update_custom_data(item_info.id, custom_data) except Exception as e: logger.info( "INFO FOR DEBUGGING", extra={ "project_id": project.id, "dataset_id": dataset.id, "names": names, "item_paths": item_paths, }, ) raise e anns_progress = None if log_progress or progress_cb is not None: anns_progress = tqdm_sly( desc="Uploading annotations to {!r}".format(dataset.name), total=len(video_ids), leave=False, ) try: api.video.annotation.upload_paths(video_ids, ann_paths, project_fs.meta, anns_progress) except Exception as e: logger.info( "INFO FOR DEBUGGING", extra={ "project_id": project.id, "dataset_id": dataset.id, "item_ids": video_ids, "ann_paths": ann_paths, }, ) raise e return project.id, project.name