:py:mod:`kwcoco` ================ .. py:module:: kwcoco .. autoapi-nested-parse:: The Kitware COCO module defines a variant of the Microsoft COCO format, originally developed for the "collected images in context" object detection challenge. We are backwards compatible with the original module, but we also have improved implementations in several places, including segmentations, keypoints, annotation tracks, multi-spectral images, and videos (which represents a generic sequence of images). A kwcoco file is a "manifest" that serves as a single reference that points to all images, categories, and annotations in a computer vision dataset. Thus, when applying an algorithm to a dataset, it is sufficient to have the algorithm take one dataset parameter: the path to the kwcoco file. Generally a kwcoco file will live in a "bundle" directory along with the data that it references, and paths in the kwcoco file will be relative to the location of the kwcoco file itself. The main data structure in this model is largely based on the implementation in https://github.com/cocodataset/cocoapi It uses the same efficient core indexing data structures, but in our implementation the indexing can be optionally turned off, functions are silent by default (with the exception of long running processes, which optionally show progress by default). We support helper functions that add and remove images, categories, and annotations. The :class:`kwcoco.CocoDataset` class is capable of dynamic addition and removal of categories, images, and annotations. Has better support for keypoints and segmentation formats than the original COCO format. Despite being written in Python, this data structure is reasonably efficient. .. code:: python >>> import kwcoco >>> import json >>> # Create demo data >>> demo = kwcoco.CocoDataset.demo() >>> # Reroot can switch between absolute / relative-paths >>> demo.reroot(absolute=True) >>> # could also use demo.dump / demo.dumps, but this is more explicit >>> text = json.dumps(demo.dataset) >>> with open('demo.json', 'w') as file: >>> file.write(text) >>> # Read from disk >>> self = kwcoco.CocoDataset('demo.json') >>> # Add data >>> cid = self.add_category('Cat') >>> gid = self.add_image('new-img.jpg') >>> aid = self.add_annotation(image_id=gid, category_id=cid, bbox=[0, 0, 100, 100]) >>> # Remove data >>> self.remove_annotations([aid]) >>> self.remove_images([gid]) >>> self.remove_categories([cid]) >>> # Look at data >>> import ubelt as ub >>> print(ub.repr2(self.basic_stats(), nl=1)) >>> print(ub.repr2(self.extended_stats(), nl=2)) >>> print(ub.repr2(self.boxsize_stats(), nl=3)) >>> print(ub.repr2(self.category_annotation_frequency())) >>> # Inspect data >>> # xdoctest: +REQUIRES(module:kwplot) >>> import kwplot >>> kwplot.autompl() >>> self.show_image(gid=1) >>> # Access single-item data via imgs, cats, anns >>> cid = 1 >>> self.cats[cid] {'id': 1, 'name': 'astronaut', 'supercategory': 'human'} >>> gid = 1 >>> self.imgs[gid] {'id': 1, 'file_name': 'astro.png', 'url': 'https://i.imgur.com/KXhKM72.png'} >>> aid = 3 >>> self.anns[aid] {'id': 3, 'image_id': 1, 'category_id': 3, 'line': [326, 369, 500, 500]} >>> # Access multi-item data via the annots and images helper objects >>> aids = self.index.gid_to_aids[2] >>> annots = self.annots(aids) >>> print('annots = {}'.format(ub.repr2(annots, nl=1, sv=1))) annots = >>> annots.lookup('category_id') [6, 4] >>> annots.lookup('bbox') [[37, 6, 230, 240], [124, 96, 45, 18]] >>> # built in conversions to efficient kwimage array DataStructures >>> print(ub.repr2(annots.detections.data, sv=1)) { 'boxes': , 'class_idxs': [5, 3], 'keypoints': , 'segmentations': , } >>> gids = list(self.imgs.keys()) >>> images = self.images(gids) >>> print('images = {}'.format(ub.repr2(images, nl=1, sv=1))) images = >>> images.lookup('file_name') ['astro.png', 'carl.png', 'stars.png'] >>> print('images.annots = {}'.format(images.annots)) images.annots = >>> print('images.annots.cids = {!r}'.format(images.annots.cids)) images.annots.cids = [[1, 2, 3, 4, 5, 5, 5, 5, 5], [6, 4], []] CocoDataset API ############### The following is a logical grouping of the public kwcoco.CocoDataset API attributes and methods. See the in-code documentation for further details. CocoDataset classmethods (via MixinCocoExtras) ********************************************** * :func:`kwcoco.CocoDataset.coerce` - Attempt to transform the input into the intended CocoDataset. * :func:`kwcoco.CocoDataset.demo` - Create a toy coco dataset for testing and demo puposes * :func:`kwcoco.CocoDataset.random` - Creates a random CocoDataset according to distribution parameters CocoDataset classmethods (via CocoDataset) ****************************************** * :func:`kwcoco.CocoDataset.from_coco_paths` - Constructor from multiple coco file paths. * :func:`kwcoco.CocoDataset.from_data` - Constructor from a json dictionary * :func:`kwcoco.CocoDataset.from_image_paths` - Constructor from a list of images paths. CocoDataset slots ***************** * :attr:`kwcoco.CocoDataset.index` - an efficient lookup index into the coco data structure. The index defines its own attributes like ``anns``, ``cats``, ``imgs``, ``gid_to_aids``, ``file_name_to_img``, etc. See :class:`CocoIndex` for more details on which attributes are available. * :attr:`kwcoco.CocoDataset.hashid` - If computed, this will be a hash uniquely identifing the dataset. To ensure this is computed see :func:`kwcoco.coco_dataset.MixinCocoExtras._build_hashid`. * :attr:`kwcoco.CocoDataset.hashid_parts` - * :attr:`kwcoco.CocoDataset.tag` - A tag indicating the name of the dataset. * :attr:`kwcoco.CocoDataset.dataset` - raw json data structure. This is the base dictionary that contains {'annotations': List, 'images': List, 'categories': List} * :attr:`kwcoco.CocoDataset.bundle_dpath` - If known, this is the root path that all image file names are relative to. This can also be manually overwritten by the user. * :attr:`kwcoco.CocoDataset.assets_dpath` - * :attr:`kwcoco.CocoDataset.cache_dpath` - CocoDataset properties ********************** * :attr:`kwcoco.CocoDataset.anns` - * :attr:`kwcoco.CocoDataset.cats` - * :attr:`kwcoco.CocoDataset.cid_to_aids` - * :attr:`kwcoco.CocoDataset.data_fpath` - * :attr:`kwcoco.CocoDataset.data_root` - * :attr:`kwcoco.CocoDataset.fpath` - if known, this stores the filepath the dataset was loaded from * :attr:`kwcoco.CocoDataset.gid_to_aids` - * :attr:`kwcoco.CocoDataset.img_root` - * :attr:`kwcoco.CocoDataset.imgs` - * :attr:`kwcoco.CocoDataset.n_annots` - * :attr:`kwcoco.CocoDataset.n_cats` - * :attr:`kwcoco.CocoDataset.n_images` - * :attr:`kwcoco.CocoDataset.n_videos` - * :attr:`kwcoco.CocoDataset.name_to_cat` - CocoDataset methods (via MixinCocoAddRemove) ******************************************** * :func:`kwcoco.CocoDataset.add_annotation` - Add an annotation to the dataset (dynamically updates the index) * :func:`kwcoco.CocoDataset.add_annotations` - Faster less-safe multi-item alternative to add_annotation. * :func:`kwcoco.CocoDataset.add_category` - Adds a category * :func:`kwcoco.CocoDataset.add_image` - Add an image to the dataset (dynamically updates the index) * :func:`kwcoco.CocoDataset.add_images` - Faster less-safe multi-item alternative * :func:`kwcoco.CocoDataset.add_video` - Add a video to the dataset (dynamically updates the index) * :func:`kwcoco.CocoDataset.clear_annotations` - Removes all annotations (but not images and categories) * :func:`kwcoco.CocoDataset.clear_images` - Removes all images and annotations (but not categories) * :func:`kwcoco.CocoDataset.ensure_category` - Like :func:`add_category`, but returns the existing category id if it already exists instead of failing. In this case all metadata is ignored. * :func:`kwcoco.CocoDataset.ensure_image` - Like :func:`add_image`,, but returns the existing image id if it already exists instead of failing. In this case all metadata is ignored. * :func:`kwcoco.CocoDataset.remove_annotation` - Remove a single annotation from the dataset * :func:`kwcoco.CocoDataset.remove_annotation_keypoints` - Removes all keypoints with a particular category * :func:`kwcoco.CocoDataset.remove_annotations` - Remove multiple annotations from the dataset. * :func:`kwcoco.CocoDataset.remove_categories` - Remove categories and all annotations in those categories. Currently does not change any hierarchy information * :func:`kwcoco.CocoDataset.remove_images` - Remove images and any annotations contained by them * :func:`kwcoco.CocoDataset.remove_keypoint_categories` - Removes all keypoints of a particular category as well as all annotation keypoints with those ids. * :func:`kwcoco.CocoDataset.remove_videos` - Remove videos and any images / annotations contained by them * :func:`kwcoco.CocoDataset.set_annotation_category` - Sets the category of a single annotation CocoDataset methods (via MixinCocoObjects) ****************************************** * :func:`kwcoco.CocoDataset.annots` - Return vectorized annotation objects * :func:`kwcoco.CocoDataset.categories` - Return vectorized category objects * :func:`kwcoco.CocoDataset.images` - Return vectorized image objects * :func:`kwcoco.CocoDataset.videos` - Return vectorized video objects CocoDataset methods (via MixinCocoStats) **************************************** * :func:`kwcoco.CocoDataset.basic_stats` - Reports number of images, annotations, and categories. * :func:`kwcoco.CocoDataset.boxsize_stats` - Compute statistics about bounding box sizes. * :func:`kwcoco.CocoDataset.category_annotation_frequency` - Reports the number of annotations of each category * :func:`kwcoco.CocoDataset.category_annotation_type_frequency` - Reports the number of annotations of each type for each category * :func:`kwcoco.CocoDataset.conform` - Make the COCO file conform a stricter spec, infers attibutes where possible. * :func:`kwcoco.CocoDataset.extended_stats` - Reports number of images, annotations, and categories. * :func:`kwcoco.CocoDataset.find_representative_images` - Find images that have a wide array of categories. Attempt to find the fewest images that cover all categories using images that contain both a large and small number of annotations. * :func:`kwcoco.CocoDataset.keypoint_annotation_frequency` - * :func:`kwcoco.CocoDataset.stats` - This function corresponds to :mod:`kwcoco.cli.coco_stats`. * :func:`kwcoco.CocoDataset.validate` - Performs checks on this coco dataset. CocoDataset methods (via MixinCocoAccessors) ******************************************** * :func:`kwcoco.CocoDataset.category_graph` - Construct a networkx category hierarchy * :func:`kwcoco.CocoDataset.delayed_load` - Experimental method * :func:`kwcoco.CocoDataset.get_auxiliary_fpath` - Returns the full path to auxiliary data for an image * :func:`kwcoco.CocoDataset.get_image_fpath` - Returns the full path to the image * :func:`kwcoco.CocoDataset.keypoint_categories` - Construct a consistent CategoryTree representation of keypoint classes * :func:`kwcoco.CocoDataset.load_annot_sample` - Reads the chip of an annotation. Note this is much less efficient than using a sampler, but it doesn't require disk cache. * :func:`kwcoco.CocoDataset.load_image` - Reads an image from disk and * :func:`kwcoco.CocoDataset.object_categories` - Construct a consistent CategoryTree representation of object classes CocoDataset methods (via CocoDataset) ************************************* * :func:`kwcoco.CocoDataset.copy` - Deep copies this object * :func:`kwcoco.CocoDataset.dump` - Writes the dataset out to the json format * :func:`kwcoco.CocoDataset.dumps` - Writes the dataset out to the json format * :func:`kwcoco.CocoDataset.subset` - Return a subset of the larger coco dataset by specifying which images to port. All annotations in those images will be taken. * :func:`kwcoco.CocoDataset.union` - Merges multiple :class:`CocoDataset` items into one. Names and associations are retained, but ids may be different. * :func:`kwcoco.CocoDataset.view_sql` - Create a cached SQL interface to this dataset suitable for large scale multiprocessing use cases. CocoDataset methods (via MixinCocoExtras) ***************************************** * :func:`kwcoco.CocoDataset.corrupted_images` - Check for images that don't exist or can't be opened * :func:`kwcoco.CocoDataset.missing_images` - Check for images that don't exist * :func:`kwcoco.CocoDataset.rename_categories` - Rename categories with a potentially coarser categorization. * :func:`kwcoco.CocoDataset.reroot` - Rebase image/data paths onto a new image/data root. CocoDataset methods (via MixinCocoDraw) *************************************** * :func:`kwcoco.CocoDataset.draw_image` - Use kwimage to draw all annotations on an image and return the pixels as a numpy array. * :func:`kwcoco.CocoDataset.imread` - Loads a particular image * :func:`kwcoco.CocoDataset.show_image` - Use matplotlib to show an image with annotations overlaid Subpackages ----------- .. toctree:: :titlesonly: :maxdepth: 3 cli/index.rst data/index.rst demo/index.rst examples/index.rst metrics/index.rst util/index.rst Submodules ---------- .. toctree:: :titlesonly: :maxdepth: 1 __main__/index.rst _helpers/index.rst abstract_coco_dataset/index.rst category_tree/index.rst channel_spec/index.rst coco_dataset/index.rst coco_evaluator/index.rst coco_image/index.rst coco_objects1d/index.rst coco_schema/index.rst coco_sql_dataset/index.rst compat_dataset/index.rst exceptions/index.rst kpf/index.rst kw18/index.rst Package Contents ---------------- Classes ~~~~~~~ .. autoapisummary:: kwcoco.AbstractCocoDataset kwcoco.CocoDataset kwcoco.CocoImage kwcoco.CategoryTree kwcoco.ChannelSpec kwcoco.FusedChannelSpec .. py:class:: AbstractCocoDataset Bases: :py:obj:`abc.ABC` This is a common base for all variants of the Coco Dataset At the time of writing there is kwcoco.CocoDataset (which is the dictionary-based backend), and the kwcoco.coco_sql_dataset.CocoSqlDataset, which is experimental. .. py:class:: CocoDataset(data=None, tag=None, bundle_dpath=None, img_root=None, fname=None, autobuild=True) Bases: :py:obj:`kwcoco.abstract_coco_dataset.AbstractCocoDataset`, :py:obj:`MixinCocoAddRemove`, :py:obj:`MixinCocoStats`, :py:obj:`MixinCocoObjects`, :py:obj:`MixinCocoDraw`, :py:obj:`MixinCocoAccessors`, :py:obj:`MixinCocoExtras`, :py:obj:`MixinCocoIndex`, :py:obj:`MixinCocoDepricate`, :py:obj:`ubelt.NiceRepr` The main coco dataset class with a json dataset backend. :ivar dataset: raw json data structure. This is the base dictionary that contains {'annotations': List, 'images': List, 'categories': List} :vartype dataset: Dict :ivar index: an efficient lookup index into the coco data structure. The index defines its own attributes like ``anns``, ``cats``, ``imgs``, ``gid_to_aids``, ``file_name_to_img``, etc. See :class:`CocoIndex` for more details on which attributes are available. :vartype index: CocoIndex :ivar fpath: if known, this stores the filepath the dataset was loaded from :vartype fpath: PathLike | None :ivar tag: A tag indicating the name of the dataset. :vartype tag: str :ivar bundle_dpath: If known, this is the root path that all image file names are relative to. This can also be manually overwritten by the user. :vartype bundle_dpath: PathLike | None :ivar hashid: If computed, this will be a hash uniquely identifing the dataset. To ensure this is computed see :func:`kwcoco.coco_dataset.MixinCocoExtras._build_hashid`. :vartype hashid: str | None .. rubric:: References http://cocodataset.org/#format http://cocodataset.org/#download .. rubric:: CommandLine .. code-block:: bash python -m kwcoco.coco_dataset CocoDataset --show .. rubric:: Example >>> from kwcoco.coco_dataset import demo_coco_data >>> import kwcoco >>> import ubelt as ub >>> # Returns a coco json structure >>> dataset = demo_coco_data() >>> # Pass the coco json structure to the API >>> self = kwcoco.CocoDataset(dataset, tag='demo') >>> # Now you can access the data using the index and helper methods >>> # >>> # Start by looking up an image by it's COCO id. >>> image_id = 1 >>> img = self.index.imgs[image_id] >>> print(ub.repr2(img, nl=1, sort=1)) { 'file_name': 'astro.png', 'id': 1, 'url': 'https://i.imgur.com/KXhKM72.png', } >>> # >>> # Use the (gid_to_aids) index to lookup annotations in the iamge >>> annotation_id = sorted(self.index.gid_to_aids[image_id])[0] >>> ann = self.index.anns[annotation_id] >>> print(ub.repr2(ub.dict_diff(ann, {'segmentation'}), nl=1)) { 'bbox': [10, 10, 360, 490], 'category_id': 1, 'id': 1, 'image_id': 1, 'keypoints': [247, 101, 2, 202, 100, 2], } >>> # >>> # Use annotation category id to look up that information >>> category_id = ann['category_id'] >>> cat = self.index.cats[category_id] >>> print('cat = {}'.format(ub.repr2(cat, nl=1, sort=1))) cat = { 'id': 1, 'name': 'astronaut', 'supercategory': 'human', } >>> # >>> # Now play with some helper functions, like extended statistics >>> extended_stats = self.extended_stats() >>> print('extended_stats = {}'.format(ub.repr2(extended_stats, nl=1, precision=2, sort=1))) extended_stats = { 'annots_per_img': {'mean': 3.67, 'std': 3.86, 'min': 0.00, 'max': 9.00, 'nMin': 1, 'nMax': 1, 'shape': (3,)}, 'imgs_per_cat': {'mean': 0.88, 'std': 0.60, 'min': 0.00, 'max': 2.00, 'nMin': 2, 'nMax': 1, 'shape': (8,)}, 'cats_per_img': {'mean': 2.33, 'std': 2.05, 'min': 0.00, 'max': 5.00, 'nMin': 1, 'nMax': 1, 'shape': (3,)}, 'annots_per_cat': {'mean': 1.38, 'std': 1.49, 'min': 0.00, 'max': 5.00, 'nMin': 2, 'nMax': 1, 'shape': (8,)}, 'imgs_per_video': {'empty_list': True}, } >>> # You can "draw" a raster of the annotated image with cv2 >>> canvas = self.draw_image(2) >>> # Or if you have matplotlib you can "show" the image with mpl objects >>> # xdoctest: +REQUIRES(--show) >>> from matplotlib import pyplot as plt >>> fig = plt.figure() >>> ax1 = fig.add_subplot(1, 2, 1) >>> self.show_image(gid=2) >>> ax2 = fig.add_subplot(1, 2, 2) >>> ax2.imshow(canvas) >>> ax1.set_title('show with matplotlib') >>> ax2.set_title('draw with cv2') >>> plt.show() .. py:method:: fpath(self) :property: In the future we will deprecate img_root for bundle_dpath .. py:method:: _infer_dirs(self) .. py:method:: from_data(CocoDataset, data, bundle_dpath=None, img_root=None) :classmethod: Constructor from a json dictionary .. py:method:: from_image_paths(CocoDataset, gpaths, bundle_dpath=None, img_root=None) :classmethod: Constructor from a list of images paths. This is a convinience method. :Parameters: **gpaths** (*List[str]*) -- list of image paths .. rubric:: Example >>> coco_dset = CocoDataset.from_image_paths(['a.png', 'b.png']) >>> assert coco_dset.n_images == 2 .. py:method:: from_coco_paths(CocoDataset, fpaths, max_workers=0, verbose=1, mode='thread', union='try') :classmethod: Constructor from multiple coco file paths. Loads multiple coco datasets and unions the result .. note:: if the union operation fails, the list of individually loaded files is returned instead. :Parameters: * **fpaths** (*List[str]*) -- list of paths to multiple coco files to be loaded and unioned. * **max_workers** (*int, default=0*) -- number of worker threads / processes * **verbose** (*int*) -- verbosity level * **mode** (*str*) -- thread, process, or serial * **union** (*str | bool, default='try'*) -- If True, unions the result datasets after loading. If False, just returns the result list. If 'try', then try to preform the union, but return the result list if it fails. .. py:method:: copy(self) Deep copies this object .. rubric:: Example >>> from kwcoco.coco_dataset import * >>> self = CocoDataset.demo() >>> new = self.copy() >>> assert new.imgs[1] is new.dataset['images'][0] >>> assert new.imgs[1] == self.dataset['images'][0] >>> assert new.imgs[1] is not self.dataset['images'][0] .. py:method:: __nice__(self) .. py:method:: dumps(self, indent=None, newlines=False) Writes the dataset out to the json format :Parameters: **newlines** (*bool*) -- if True, each annotation, image, category gets its own line .. note:: Using newlines=True is similar to: print(ub.repr2(dset.dataset, nl=2, trailsep=False)) However, the above may not output valid json if it contains ndarrays. .. rubric:: Example >>> from kwcoco.coco_dataset import * >>> import json >>> self = CocoDataset.demo() >>> text = self.dumps(newlines=True) >>> print(text) >>> self2 = CocoDataset(json.loads(text), tag='demo2') >>> assert self2.dataset == self.dataset >>> assert self2.dataset is not self.dataset >>> text = self.dumps(newlines=True) >>> print(text) >>> self2 = CocoDataset(json.loads(text), tag='demo2') >>> assert self2.dataset == self.dataset >>> assert self2.dataset is not self.dataset .. rubric:: Example >>> from kwcoco.coco_dataset import * >>> self = CocoDataset.coerce('vidshapes1-msi-multisensor', verbose=3) >>> self.remove_annotations(self.annots()) >>> text = self.dumps(newlines=True, indent=' ') >>> print(text) .. py:method:: dump(self, file, indent=None, newlines=False) Writes the dataset out to the json format :Parameters: * **file** (*PathLike | FileLike*) -- Where to write the data. Can either be a path to a file or an open file pointer / stream. * **newlines** (*bool*) -- if True, each annotation, image, category gets its own line. .. rubric:: Example >>> import tempfile >>> from kwcoco.coco_dataset import * >>> self = CocoDataset.demo() >>> file = tempfile.NamedTemporaryFile('w') >>> self.dump(file) >>> file.seek(0) >>> text = open(file.name, 'r').read() >>> print(text) >>> file.seek(0) >>> dataset = json.load(open(file.name, 'r')) >>> self2 = CocoDataset(dataset, tag='demo2') >>> assert self2.dataset == self.dataset >>> assert self2.dataset is not self.dataset >>> file = tempfile.NamedTemporaryFile('w') >>> self.dump(file, newlines=True) >>> file.seek(0) >>> text = open(file.name, 'r').read() >>> print(text) >>> file.seek(0) >>> dataset = json.load(open(file.name, 'r')) >>> self2 = CocoDataset(dataset, tag='demo2') >>> assert self2.dataset == self.dataset >>> assert self2.dataset is not self.dataset .. py:method:: _check_json_serializable(self, verbose=1) Debug which part of a coco dataset might not be json serializable .. py:method:: _check_integrity(self) perform all checks .. py:method:: _check_index(self) .. rubric:: Example >>> import kwcoco >>> self = kwcoco.CocoDataset.demo() >>> self._check_index() >>> # Force a failure >>> self.index.anns.pop(1) >>> self.index.anns.pop(2) >>> import pytest >>> with pytest.raises(AssertionError): >>> self._check_index() .. py:method:: _check_pointers(self, verbose=1) Check that all category and image ids referenced by annotations exist .. py:method:: _build_index(self) .. py:method:: union(*others, disjoint_tracks=True, **kwargs) Merges multiple :class:`CocoDataset` items into one. Names and associations are retained, but ids may be different. :Parameters: * **\*others** -- a series of CocoDatasets that we will merge. Note, if called as an instance method, the "self" instance will be the first item in the "others" list. But if called like a classmethod, "others" will be empty by default. * **disjoint_tracks** (*bool, default=True*) -- if True, we will assume track-ids are disjoint and if two datasets share the same track-id, we will disambiguate them. Otherwise they will be copied over as-is. * **\*\*kwargs** -- constructor options for the new merged CocoDataset :returns: a new merged coco dataset :rtype: CocoDataset .. rubric:: CommandLine .. code-block:: bash xdoctest -m kwcoco.coco_dataset CocoDataset.union .. rubric:: Example >>> # Test union works with different keypoint categories >>> dset1 = CocoDataset.demo('shapes1') >>> dset2 = CocoDataset.demo('shapes2') >>> dset1.remove_keypoint_categories(['bot_tip', 'mid_tip', 'right_eye']) >>> dset2.remove_keypoint_categories(['top_tip', 'left_eye']) >>> dset_12a = CocoDataset.union(dset1, dset2) >>> dset_12b = dset1.union(dset2) >>> dset_21 = dset2.union(dset1) >>> def add_hist(h1, h2): >>> return {k: h1.get(k, 0) + h2.get(k, 0) for k in set(h1) | set(h2)} >>> kpfreq1 = dset1.keypoint_annotation_frequency() >>> kpfreq2 = dset2.keypoint_annotation_frequency() >>> kpfreq_want = add_hist(kpfreq1, kpfreq2) >>> kpfreq_got1 = dset_12a.keypoint_annotation_frequency() >>> kpfreq_got2 = dset_12b.keypoint_annotation_frequency() >>> assert kpfreq_want == kpfreq_got1 >>> assert kpfreq_want == kpfreq_got2 >>> # Test disjoint gid datasets >>> import kwcoco >>> dset1 = kwcoco.CocoDataset.demo('shapes3') >>> for new_gid, img in enumerate(dset1.dataset['images'], start=10): >>> for aid in dset1.gid_to_aids[img['id']]: >>> dset1.anns[aid]['image_id'] = new_gid >>> img['id'] = new_gid >>> dset1.index.clear() >>> dset1._build_index() >>> # ------ >>> dset2 = kwcoco.CocoDataset.demo('shapes2') >>> for new_gid, img in enumerate(dset2.dataset['images'], start=100): >>> for aid in dset2.gid_to_aids[img['id']]: >>> dset2.anns[aid]['image_id'] = new_gid >>> img['id'] = new_gid >>> dset1.index.clear() >>> dset2._build_index() >>> others = [dset1, dset2] >>> merged = kwcoco.CocoDataset.union(*others) >>> print('merged = {!r}'.format(merged)) >>> print('merged.imgs = {}'.format(ub.repr2(merged.imgs, nl=1))) >>> assert set(merged.imgs) & set([10, 11, 12, 100, 101]) == set(merged.imgs) >>> # Test data is not preserved >>> dset2 = kwcoco.CocoDataset.demo('shapes2') >>> dset1 = kwcoco.CocoDataset.demo('shapes3') >>> others = (dset1, dset2) >>> cls = self = kwcoco.CocoDataset >>> merged = cls.union(*others) >>> print('merged = {!r}'.format(merged)) >>> print('merged.imgs = {}'.format(ub.repr2(merged.imgs, nl=1))) >>> assert set(merged.imgs) & set([1, 2, 3, 4, 5]) == set(merged.imgs) >>> # Test track-ids are mapped correctly >>> dset1 = kwcoco.CocoDataset.demo('vidshapes1') >>> dset2 = kwcoco.CocoDataset.demo('vidshapes2') >>> dset3 = kwcoco.CocoDataset.demo('vidshapes3') >>> others = (dset1, dset2, dset3) >>> for dset in others: >>> [a.pop('segmentation', None) for a in dset.index.anns.values()] >>> [a.pop('keypoints', None) for a in dset.index.anns.values()] >>> cls = self = kwcoco.CocoDataset >>> merged = cls.union(*others, disjoint_tracks=1) >>> print('dset1.anns = {}'.format(ub.repr2(dset1.anns, nl=1))) >>> print('dset2.anns = {}'.format(ub.repr2(dset2.anns, nl=1))) >>> print('dset3.anns = {}'.format(ub.repr2(dset3.anns, nl=1))) >>> print('merged.anns = {}'.format(ub.repr2(merged.anns, nl=1))) .. rubric:: Example >>> import kwcoco >>> # Test empty union >>> empty_union = kwcoco.CocoDataset.union() >>> assert len(empty_union.index.imgs) == 0 .. todo:: - [ ] are supercategories broken? - [ ] reuse image ids where possible - [ ] reuse annotation / category ids where possible - [X] handle case where no inputs are given - [x] disambiguate track-ids - [x] disambiguate video-ids .. py:method:: subset(self, gids, copy=False, autobuild=True) Return a subset of the larger coco dataset by specifying which images to port. All annotations in those images will be taken. :Parameters: * **gids** (*List[int]*) -- image-ids to copy into a new dataset * **copy** (*bool, default=False*) -- if True, makes a deep copy of all nested attributes, otherwise makes a shallow copy. * **autobuild** (*bool, default=True*) -- if True will automatically build the fast lookup index. .. rubric:: Example >>> self = CocoDataset.demo() >>> gids = [1, 3] >>> sub_dset = self.subset(gids) >>> assert len(self.index.gid_to_aids) == 3 >>> assert len(sub_dset.gid_to_aids) == 2 .. rubric:: Example >>> import kwcoco >>> self = kwcoco.CocoDataset.demo('vidshapes2') >>> gids = [1, 2] >>> sub_dset = self.subset(gids, copy=True) >>> assert len(sub_dset.index.videos) == 1 >>> assert len(self.index.videos) == 2 .. rubric:: Example >>> self = CocoDataset.demo() >>> sub1 = self.subset([1]) >>> sub2 = self.subset([2]) >>> sub3 = self.subset([3]) >>> others = [sub1, sub2, sub3] >>> rejoined = CocoDataset.union(*others) >>> assert len(sub1.anns) == 9 >>> assert len(sub2.anns) == 2 >>> assert len(sub3.anns) == 0 >>> assert rejoined.basic_stats() == self.basic_stats() .. py:method:: view_sql(self, force_rewrite=False, memory=False) Create a cached SQL interface to this dataset suitable for large scale multiprocessing use cases. :Parameters: * **force_rewrite** (*bool, default=False*) -- if True, forces an update to any existing cache file on disk * **memory** (*bool, default=False*) -- if True, the database is constructed in memory. .. note:: This view cache is experimental and currently depends on the timestamp of the file pointed to by ``self.fpath``. In other words dont use this on in-memory datasets. .. py:class:: CocoImage(img, dset=None) Bases: :py:obj:`ubelt.NiceRepr` An object-oriented representation of a coco image. It provides helper methods that are specific to a single image. This operates directly on a single coco image dictionary, but it can optionally be connected to a parent dataset, which allows it to use CocoDataset methods to query about relationships and resolve pointers. This is different than the Images class in coco_object1d, which is just a vectorized interface to multiple objects. .. rubric:: Example >>> import kwcoco >>> dset1 = kwcoco.CocoDataset.demo('shapes8') >>> dset2 = kwcoco.CocoDataset.demo('vidshapes8-multispectral') >>> self = CocoImage(dset1.imgs[1], dset1) >>> print('self = {!r}'.format(self)) >>> print('self.channels = {}'.format(ub.repr2(self.channels, nl=1))) >>> self = CocoImage(dset2.imgs[1], dset2) >>> print('self.channels = {}'.format(ub.repr2(self.channels, nl=1))) >>> self.primary_asset() .. py:method:: from_gid(cls, dset, gid) :classmethod: .. py:method:: bundle_dpath(self) :property: .. py:method:: video(self) :property: Helper to grab the video for this image if it exists .. py:method:: detach(self) Removes references to the underlying coco dataset, but keeps special information such that it wont be needed. .. py:method:: __nice__(self) .. rubric:: Example >>> from kwcoco.coco_image import * # NOQA >>> import kwcoco >>> with ub.CaptureStdout() as cap: ... dset = kwcoco.CocoDataset.demo('shapes8') >>> self = CocoImage(dset.dataset['images'][0], dset) >>> print('self = {!r}'.format(self)) >>> dset = kwcoco.CocoDataset.demo() >>> self = CocoImage(dset.dataset['images'][0], dset) >>> print('self = {!r}'.format(self)) .. py:method:: stats(self) .. py:method:: __getitem__(self, key) Proxy getter attribute for underlying `self.img` dictionary .. py:method:: keys(self) Proxy getter attribute for underlying `self.img` dictionary .. py:method:: get(self, key, default=ub.NoParam) Proxy getter attribute for underlying `self.img` dictionary .. py:method:: channels(self) :property: .. py:method:: num_channels(self) :property: .. py:method:: dsize(self) :property: .. py:method:: primary_image_filepath(self, requires=None) .. py:method:: primary_asset(self, requires=None) Compute a "main" image asset. .. rubric:: Notes Uses a heuristic. * First, try to find the auxiliary image that has with the smallest distortion to the base image (if known via warp_aux_to_img) * Second, break ties by using the largest image if w / h is known * Last, if previous information not available use the first auxiliary image. :Parameters: **requires** (*List[str]*) -- list of attribute that must be non-None to consider an object as the primary one. .. todo:: - [ ] Add in primary heuristics .. rubric:: Example >>> import kwarray >>> from kwcoco.coco_image import * # NOQA >>> rng = kwarray.ensure_rng(0) >>> def random_auxiliary(name, w=None, h=None): >>> return {'file_name': name, 'width': w, 'height': h} >>> self = CocoImage({ >>> 'auxiliary': [ >>> random_auxiliary('1'), >>> random_auxiliary('2'), >>> random_auxiliary('3'), >>> ] >>> }) >>> assert self.primary_asset()['file_name'] == '1' >>> self = CocoImage({ >>> 'auxiliary': [ >>> random_auxiliary('1'), >>> random_auxiliary('2', 3, 3), >>> random_auxiliary('3'), >>> ] >>> }) >>> assert self.primary_asset()['file_name'] == '2' .. py:method:: iter_image_filepaths(self) .. py:method:: iter_asset_objs(self) Iterate through base + auxiliary dicts that have file paths :Yields: *dict* -- an image or auxiliary dictionary .. py:method:: find_asset_obj(self, channels) Find the asset dictionary with the specified channels .. py:method:: add_auxiliary_item(self, file_name=None, channels=None, imdata=None, warp_aux_to_img=None, width=None, height=None, imwrite=False) Adds an auxiliary item to the image dictionary. This operation can be done purely in-memory (the default), or the image data can be written to a file on disk (via the imwrite=True flag). :Parameters: * **file_name** (*str | None*) -- The name of the file relative to the bundle directory. If unspecified, imdata must be given. * **channels** (*str | kwcoco.FusedChannelSpec*) -- The channel code indicating what each of the bands represents. These channels should be disjoint wrt to the existing data in this image (this is not checked). * **imdata** (*ndarray | None*) -- The underlying image data this auxiliary item represents. If unspecified, it is assumed file_name points to a path on disk that will eventually exist. If imdata, file_name, and the special imwrite=True flag are specified, this function will write the data to disk. * **warp_aux_to_img** (*kwimage.Affine*) -- The transformation from this auxiliary space to image space. If unspecified, assumes this item is related to image space by only a scale factor. * **width** (*int*) -- Width of the data in auxiliary space (inferred if unspecified) * **height** (*int*) -- Height of the data in auxiliary space (inferred if unspecified) * **imwrite** (*bool*) -- If specified, both imdata and file_name must be specified, and this will write the data to disk. Note: it it recommended that you simply call imwrite yourself before or after calling this function. This lets you better control imwrite parameters. .. todo:: - [ ] Allow imwrite to specify an executor that is used to return a Future so the imwrite call does not block. .. rubric:: Example >>> from kwcoco.coco_image import * # NOQA >>> import kwcoco >>> dset = kwcoco.CocoDataset.demo('vidshapes8-multispectral') >>> coco_img = dset.coco_image(1) >>> imdata = np.random.rand(32, 32, 5) >>> channels = kwcoco.FusedChannelSpec.coerce('Aux:5') >>> coco_img.add_auxiliary_item(imdata=imdata, channels=channels) .. py:method:: delay(self, channels=None, space='image', bundle_dpath=None) Perform a delayed load on the data in this image. The delayed load can load a subset of channels, and perform lazy warping operations. If the underlying data is in a tiled format this can reduce the amount of disk IO needed to read the data if only a small crop or lower resolution view of the data is needed. .. note:: This method is experimental and relies on the delayed load proof-of-concept. :Parameters: * **gid** (*int*) -- image id to load * **channels** (*FusedChannelSpec*) -- specific channels to load. if unspecified, all channels are loaded. * **space** (*str*) -- can either be "image" for loading in image space, or "video" for loading in video space. .. todo:: - [X] Currently can only take all or none of the channels from each base-image / auxiliary dict. For instance if the main image is r|g|b you can't just select g|b at the moment. - [X] The order of the channels in the delayed load should match the requested channel order. - [X] TODO: add nans to bands that don't exist or throw an error - [ ] This function could stand to have a better name. Maybe imread with a delayed=True flag? Or maybe just delayed_load? .. rubric:: Example >>> from kwcoco.coco_image import * # NOQA >>> import kwcoco >>> gid = 1 >>> # >>> dset = kwcoco.CocoDataset.demo('vidshapes8-multispectral') >>> self = CocoImage(dset.imgs[gid], dset) >>> delayed = self.delay() >>> print('delayed = {!r}'.format(delayed)) >>> print('delayed.finalize() = {!r}'.format(delayed.finalize())) >>> print('delayed.finalize() = {!r}'.format(delayed.finalize(as_xarray=True))) >>> # >>> dset = kwcoco.CocoDataset.demo('shapes8') >>> delayed = dset.delayed_load(gid) >>> print('delayed = {!r}'.format(delayed)) >>> print('delayed.finalize() = {!r}'.format(delayed.finalize())) >>> print('delayed.finalize() = {!r}'.format(delayed.finalize(as_xarray=True))) >>> crop = delayed.delayed_crop((slice(0, 3), slice(0, 3))) >>> crop.finalize() >>> crop.finalize(as_xarray=True) >>> # TODO: should only select the "red" channel >>> dset = kwcoco.CocoDataset.demo('shapes8') >>> delayed = CocoImage(dset.imgs[gid], dset).delay(channels='r') >>> import kwcoco >>> gid = 1 >>> # >>> dset = kwcoco.CocoDataset.demo('vidshapes8-multispectral') >>> delayed = dset.delayed_load(gid, channels='B1|B2', space='image') >>> print('delayed = {!r}'.format(delayed)) >>> print('delayed.finalize() = {!r}'.format(delayed.finalize(as_xarray=True))) >>> delayed = dset.delayed_load(gid, channels='B1|B2|B11', space='image') >>> print('delayed = {!r}'.format(delayed)) >>> print('delayed.finalize() = {!r}'.format(delayed.finalize(as_xarray=True))) >>> delayed = dset.delayed_load(gid, channels='B8|B1', space='video') >>> print('delayed = {!r}'.format(delayed)) >>> print('delayed.finalize() = {!r}'.format(delayed.finalize(as_xarray=True))) >>> delayed = dset.delayed_load(gid, channels='B8|foo|bar|B1', space='video') >>> print('delayed = {!r}'.format(delayed)) >>> print('delayed.finalize() = {!r}'.format(delayed.finalize(as_xarray=True))) .. rubric:: Example >>> import kwcoco >>> dset = kwcoco.CocoDataset.demo() >>> coco_img = dset.coco_image(1) >>> # Test case where nothing is registered in the dataset >>> delayed = coco_img.delay() >>> final = delayed.finalize() >>> assert final.shape == (512, 512, 3) .. rubric:: Example >>> # Test that delay works when imdata is stored in the image >>> # dictionary itself. >>> from kwcoco.coco_image import * # NOQA >>> import kwcoco >>> dset = kwcoco.CocoDataset.demo('vidshapes8-multispectral') >>> coco_img = dset.coco_image(1) >>> imdata = np.random.rand(6, 6, 5) >>> imdata[:] = np.arange(5)[None, None, :] >>> channels = kwcoco.FusedChannelSpec.coerce('Aux:5') >>> coco_img.add_auxiliary_item(imdata=imdata, channels=channels) >>> delayed = coco_img.delay(channels='B1|Aux:2:4') >>> final = delayed.finalize() .. rubric:: Example >>> # Test delay when loading in auxiliary space >>> from kwcoco.coco_image import * # NOQA >>> import kwcoco >>> dset = kwcoco.CocoDataset.demo('vidshapes8-msi-multisensor') >>> coco_img = dset.coco_image(1) >>> stream1 = coco_img.channels.streams()[0] >>> stream2 = coco_img.channels.streams()[1] >>> aux_delayed = coco_img.delay(stream1, space='auxiliary') >>> img_delayed = coco_img.delay(stream1, space='image') >>> vid_delayed = coco_img.delay(stream1, space='video') >>> # >>> aux_imdata = aux_delayed.finalize() >>> img_imdata = img_delayed.finalize() >>> assert aux_imdata.shape != img_imdata.shape >>> # Cannot load multiple auxiliary items at the same time in >>> # auxiliary space >>> import pytest >>> fused_channels = stream1 | stream2 >>> with pytest.raises(kwcoco.exceptions.CoordinateCompatibilityError): >>> aux_delayed2 = coco_img.delay(fused_channels, space='auxiliary') .. py:method:: valid_region(self, space='image') If this image has a valid polygon, return it in image, or video space .. py:method:: warp_vid_from_img(self) .. py:method:: warp_img_from_vid(self) .. py:method:: _annot_segmentation(self, ann, space='video') .. py:class:: CategoryTree(graph=None, checks=True) Bases: :py:obj:`ubelt.NiceRepr` Wrapper that maintains flat or hierarchical category information. Helps compute softmaxes and probabilities for tree-based categories where a directed edge (A, B) represents that A is a superclass of B. .. note:: There are three basic properties that this object maintains: .. code:: node: Alphanumeric string names that should be generally descriptive. Using spaces and special characters in these names is discouraged, but can be done. This is the COCO category "name" attribute. For categories this may be denoted as (name, node, cname, catname). id: The integer id of a category should ideally remain consistent. These are often given by a dataset (e.g. a COCO dataset). This is the COCO category "id" attribute. For categories this is often denoted as (id, cid). index: Contigous zero-based indices that indexes the list of categories. These should be used for the fastest access in backend computation tasks. Typically corresponds to the ordering of the channels in the final linear layer in an associated model. For categories this is often denoted as (index, cidx, idx, or cx). :ivar idx_to_node: a list of class names. Implicitly maps from index to category name. :vartype idx_to_node: List[str] :ivar id_to_node: maps integer ids to category names :vartype id_to_node: Dict[int, str] :ivar node_to_id: maps category names to ids :vartype node_to_id: Dict[str, int] :ivar node_to_idx: maps category names to indexes :vartype node_to_idx: Dict[str, int] :ivar graph: a Graph that stores any hierarchy information. For standard mutually exclusive classes, this graph is edgeless. Nodes in this graph can maintain category attributes / properties. :vartype graph: networkx.Graph :ivar idx_groups: groups of category indices that share the same parent category. :vartype idx_groups: List[List[int]] .. rubric:: Example >>> from kwcoco.category_tree import * >>> graph = nx.from_dict_of_lists({ >>> 'background': [], >>> 'foreground': ['animal'], >>> 'animal': ['mammal', 'fish', 'insect', 'reptile'], >>> 'mammal': ['dog', 'cat', 'human', 'zebra'], >>> 'zebra': ['grevys', 'plains'], >>> 'grevys': ['fred'], >>> 'dog': ['boxer', 'beagle', 'golden'], >>> 'cat': ['maine coon', 'persian', 'sphynx'], >>> 'reptile': ['bearded dragon', 't-rex'], >>> }, nx.DiGraph) >>> self = CategoryTree(graph) >>> print(self) .. rubric:: Example >>> # The coerce classmethod is the easiest way to create an instance >>> import kwcoco >>> kwcoco.CategoryTree.coerce(['a', 'b', 'c']) >> kwcoco.CategoryTree.coerce(4) >> kwcoco.CategoryTree.coerce(4) .. py:method:: copy(self) .. py:method:: from_mutex(cls, nodes, bg_hack=True) :classmethod: :Parameters: **nodes** (*List[str]*) -- or a list of class names (in which case they will all be assumed to be mutually exclusive) .. rubric:: Example >>> print(CategoryTree.from_mutex(['a', 'b', 'c'])) .. py:method:: from_json(cls, state) :classmethod: :Parameters: **state** (*Dict*) -- see __getstate__ / __json__ for details .. py:method:: from_coco(cls, categories) :classmethod: Create a CategoryTree object from coco categories :Parameters: **List[Dict]** -- list of coco-style categories .. py:method:: coerce(cls, data, **kw) :classmethod: Attempt to coerce data as a CategoryTree object. This is primarily useful for when the software stack depends on categories being represent This will work if the input data is a specially formatted json dict, a list of mutually exclusive classes, or if it is already a CategoryTree. Otherwise an error will be thrown. :Parameters: * **data** (*object*) -- a known representation of a category tree. * **\*\*kwargs** -- input type specific arguments :returns: self :rtype: CategoryTree :raises TypeError - if the input format is unknown: :raises ValueError - if kwargs are not compatible with the input format: .. rubric:: Example >>> import kwcoco >>> classes1 = kwcoco.CategoryTree.coerce(3) # integer >>> classes2 = kwcoco.CategoryTree.coerce(classes1.__json__()) # graph dict >>> classes3 = kwcoco.CategoryTree.coerce(['class_1', 'class_2', 'class_3']) # mutex list >>> classes4 = kwcoco.CategoryTree.coerce(classes1.graph) # nx Graph >>> classes5 = kwcoco.CategoryTree.coerce(classes1) # cls >>> # xdoctest: +REQUIRES(module:ndsampler) >>> import ndsampler >>> classes6 = ndsampler.CategoryTree.coerce(3) >>> classes7 = ndsampler.CategoryTree.coerce(classes1) >>> classes8 = kwcoco.CategoryTree.coerce(classes6) .. py:method:: demo(cls, key='coco', **kwargs) :classmethod: :Parameters: **key** (*str*) -- specify which demo dataset to use. Can be 'coco' (which uses the default coco demo data). Can be 'btree' which creates a binary tree and accepts kwargs 'r' and 'h' for branching-factor and height. Can be 'btree2', which is the same as btree but returns strings .. rubric:: CommandLine .. code-block:: bash xdoctest -m ~/code/kwcoco/kwcoco/category_tree.py CategoryTree.demo .. rubric:: Example >>> from kwcoco.category_tree import * >>> self = CategoryTree.demo() >>> print('self = {}'.format(self)) self = .. py:method:: to_coco(self) Converts to a coco-style data structure :Yields: *Dict* -- coco category dictionaries .. py:method:: id_to_idx(self) .. rubric:: Example >>> import kwcoco >>> self = kwcoco.CategoryTree.demo() >>> self.id_to_idx[1] .. py:method:: idx_to_id(self) .. rubric:: Example >>> import kwcoco >>> self = kwcoco.CategoryTree.demo() >>> self.idx_to_id[0] .. py:method:: idx_to_ancestor_idxs(self, include_self=True) Mapping from a class index to its ancestors :Parameters: **include_self** (*bool, default=True*) -- if True includes each node as its own ancestor. .. py:method:: idx_to_descendants_idxs(self, include_self=False) Mapping from a class index to its descendants (including itself) :Parameters: **include_self** (*bool, default=False*) -- if True includes each node as its own descendant. .. py:method:: idx_pairwise_distance(self) Get a matrix encoding the distance from one class to another. Distances * from parents to children are positive (descendants), * from children to parents are negative (ancestors), * between unreachable nodes (wrt to forward and reverse graph) are nan. .. py:method:: __len__(self) .. py:method:: __iter__(self) .. py:method:: __getitem__(self, index) .. py:method:: __contains__(self, node) .. py:method:: __json__(self) .. rubric:: Example >>> import pickle >>> self = CategoryTree.demo() >>> print('self = {!r}'.format(self.__json__())) .. py:method:: __getstate__(self) Serializes information in this class .. rubric:: Example >>> from kwcoco.category_tree import * >>> import pickle >>> self = CategoryTree.demo() >>> state = self.__getstate__() >>> serialization = pickle.dumps(self) >>> recon = pickle.loads(serialization) >>> assert recon.__json__() == self.__json__() .. py:method:: __setstate__(self, state) .. py:method:: __nice__(self) .. py:method:: is_mutex(self) Returns True if all categories are mutually exclusive (i.e. flat) If true, then the classes may be represented as a simple list of class names without any loss of information, otherwise the underlying category graph is necessary to preserve all knowledge. .. todo:: - [ ] what happens when we have a dummy root? .. py:method:: num_classes(self) :property: .. py:method:: class_names(self) :property: .. py:method:: category_names(self) :property: .. py:method:: cats(self) :property: Returns a mapping from category names to category attributes. If this category tree was constructed from a coco-dataset, then this will contain the coco category attributes. :returns: Dict[str, Dict[str, object]] .. rubric:: Example >>> from kwcoco.category_tree import * >>> self = CategoryTree.demo() >>> print('self.cats = {!r}'.format(self.cats)) .. py:method:: index(self, node) Return the index that corresponds to the category name .. py:method:: _build_index(self) construct lookup tables .. py:method:: show(self) .. py:method:: forest_str(self) .. py:method:: normalize(self) Applies a normalization scheme to the categories. Note: this may break other tasks that depend on exact category names. :returns: CategoryTree .. rubric:: Example >>> from kwcoco.category_tree import * # NOQA >>> import kwcoco >>> orig = kwcoco.CategoryTree.demo('animals_v1') >>> self = kwcoco.CategoryTree(nx.relabel_nodes(orig.graph, str.upper)) >>> norm = self.normalize() .. py:class:: ChannelSpec(spec, parsed=None) Bases: :py:obj:`BaseChannelSpec` Parse and extract information about network input channel specs for early or late fusion networks. Behaves like a dictionary of FusedChannelSpec objects .. todo:: - [ ] Rename to something that indicates this is a collection of FusedChannelSpec? MultiChannelSpec? .. note:: This class name and API is in flux and subject to change. .. note:: The pipe ('|') character represents an early-fused input stream, and order matters (it is non-communative). The comma (',') character separates different inputs streams/branches for a multi-stream/branch network which will be lated fused. Order does not matter .. rubric:: Example >>> from kwcoco.channel_spec import * # NOQA >>> # Integer spec >>> ChannelSpec.coerce(3) >>> # single mode spec >>> ChannelSpec.coerce('rgb') >>> # early fused input spec >>> ChannelSpec.coerce('rgb|disprity') >>> # late fused input spec >>> ChannelSpec.coerce('rgb,disprity') >>> # early and late fused input spec >>> ChannelSpec.coerce('rgb|ir,disprity') .. rubric:: Example >>> self = ChannelSpec('gray') >>> print('self.info = {}'.format(ub.repr2(self.info, nl=1))) >>> self = ChannelSpec('rgb') >>> print('self.info = {}'.format(ub.repr2(self.info, nl=1))) >>> self = ChannelSpec('rgb|disparity') >>> print('self.info = {}'.format(ub.repr2(self.info, nl=1))) >>> self = ChannelSpec('rgb|disparity,disparity') >>> print('self.info = {}'.format(ub.repr2(self.info, nl=1))) >>> self = ChannelSpec('rgb,disparity,flowx|flowy') >>> print('self.info = {}'.format(ub.repr2(self.info, nl=1))) .. rubric:: Example >>> specs = [ >>> 'rgb', # and rgb input >>> 'rgb|disprity', # rgb early fused with disparity >>> 'rgb,disprity', # rgb early late with disparity >>> 'rgb|ir,disprity', # rgb early fused with ir and late fused with disparity >>> 3, # 3 unknown channels >>> ] >>> for spec in specs: >>> print('=======================') >>> print('spec = {!r}'.format(spec)) >>> # >>> self = ChannelSpec.coerce(spec) >>> print('self = {!r}'.format(self)) >>> sizes = self.sizes() >>> print('sizes = {!r}'.format(sizes)) >>> print('self.info = {}'.format(ub.repr2(self.info, nl=1))) >>> # >>> item = self._demo_item((1, 1), rng=0) >>> inputs = self.encode(item) >>> components = self.decode(inputs) >>> input_shapes = ub.map_vals(lambda x: x.shape, inputs) >>> component_shapes = ub.map_vals(lambda x: x.shape, components) >>> print('item = {}'.format(ub.repr2(item, precision=1))) >>> print('inputs = {}'.format(ub.repr2(inputs, precision=1))) >>> print('input_shapes = {}'.format(ub.repr2(input_shapes))) >>> print('components = {}'.format(ub.repr2(components, precision=1))) >>> print('component_shapes = {}'.format(ub.repr2(component_shapes, nl=1))) .. py:method:: spec(self) :property: The string encodeing of this spec :returns: str .. py:method:: __contains__(self, key) .. rubric:: Example >>> 'disparity' in ChannelSpec('rgb,disparity,flowx|flowy') True >>> 'gray' in ChannelSpec('rgb,disparity,flowx|flowy') False .. py:method:: info(self) :property: .. py:method:: coerce(cls, data) :classmethod: Attempt to interpret the data as a channel specification :returns: ChannelSpec .. rubric:: Example >>> from kwcoco.channel_spec import * # NOQA >>> data = FusedChannelSpec.coerce(3) >>> assert ChannelSpec.coerce(data).spec == 'u0|u1|u2' >>> data = ChannelSpec.coerce(3) >>> assert data.spec == 'u0|u1|u2' >>> assert ChannelSpec.coerce(data).spec == 'u0|u1|u2' >>> data = ChannelSpec.coerce('u:3') >>> assert data.normalize().spec == 'u.0|u.1|u.2' .. py:method:: parse(self) Build internal representation .. rubric:: Example >>> from kwcoco.channel_spec import * # NOQA >>> self = ChannelSpec('b1|b2|b3|rgb,B:3') >>> print(self.parse()) >>> print(self.normalize().parse()) >>> ChannelSpec('').parse() .. rubric:: Example >>> base = ChannelSpec('rgb|disparity,flowx|r|flowy') >>> other = ChannelSpec('rgb') >>> self = base.intersection(other) >>> assert self.numel() == 4 .. py:method:: concise(self) .. rubric:: Example >>> self = ChannelSpec('b1|b2,b3|rgb|B.0,B.1|B.2') >>> print(self.concise().spec) b1|b2,b3|r|g|b|B.0,B.1:3 .. py:method:: normalize(self) Replace aliases with explicit single-band-per-code specs :returns: normalized spec :rtype: ChannelSpec .. rubric:: Example >>> self = ChannelSpec('b1|b2,b3|rgb,B:3') >>> normed = self.normalize() >>> print('self = {}'.format(self)) >>> print('normed = {}'.format(normed)) self = normed = .. py:method:: keys(self) .. py:method:: values(self) .. py:method:: items(self) .. py:method:: fuse(self) Fuse all parts into an early fused channel spec :returns: FusedChannelSpec .. rubric:: Example >>> from kwcoco.channel_spec import * # NOQA >>> self = ChannelSpec.coerce('b1|b2,b3|rgb,B:3') >>> fused = self.fuse() >>> print('self = {}'.format(self)) >>> print('fused = {}'.format(fused)) self = fused = .. py:method:: streams(self) Breaks this spec up into one spec for each early-fused input stream .. rubric:: Example self = ChannelSpec.coerce('r|g,B1|B2,fx|fy') list(map(len, self.streams())) .. py:method:: code_list(self) .. py:method:: as_path(self) Returns a string suitable for use in a path. Note, this may no longer be a valid channel spec .. py:method:: difference(self, other) Set difference. Remove all instances of other channels from this set of channels. .. rubric:: Example >>> from kwcoco.channel_spec import * >>> self = ChannelSpec('rgb|disparity,flowx|r|flowy') >>> other = ChannelSpec('rgb') >>> print(self.difference(other)) >>> other = ChannelSpec('flowx') >>> print(self.difference(other)) .. rubric:: Example >>> from kwcoco.channel_spec import * >>> self = ChannelSpec('a|b,c|d') >>> new = self - {'a', 'b'} >>> len(new.sizes()) == 1 >>> empty = new - 'c|d' >>> assert empty.numel() == 0 .. py:method:: intersection(self, other) Set difference. Remove all instances of other channels from this set of channels. .. rubric:: Example >>> from kwcoco.channel_spec import * >>> self = ChannelSpec('rgb|disparity,flowx|r|flowy') >>> other = ChannelSpec('rgb') >>> new = self.intersection(other) >>> print(new) >>> print(new.numel()) >>> other = ChannelSpec('flowx') >>> new = self.intersection(other) >>> print(new) >>> print(new.numel()) 4 1 .. py:method:: union(self, other) Union simply tags on a second channel spec onto this one. Duplicates are maintained. .. rubric:: Example >>> from kwcoco.channel_spec import * >>> self = ChannelSpec('rgb|disparity,flowx|r|flowy') >>> other = ChannelSpec('rgb') >>> new = self.union(other) >>> print(new) >>> print(new.numel()) >>> other = ChannelSpec('flowx') >>> new = self.union(other) >>> print(new) >>> print(new.numel()) 10 8 .. py:method:: issubset(self, other) :abstractmethod: .. py:method:: issuperset(self, other) :abstractmethod: .. py:method:: numel(self) Total number of channels in this spec .. py:method:: sizes(self) Number of dimensions for each fused stream channel IE: The EARLY-FUSED channel sizes .. rubric:: Example >>> self = ChannelSpec('rgb|disparity,flowx|flowy,B:10') >>> self.normalize().concise() >>> self.sizes() .. py:method:: unique(self, normalize=False) Returns the unique channels that will need to be given or loaded .. py:method:: _item_shapes(self, dims) Expected shape for an input item :Parameters: **dims** (*Tuple[int, int]*) -- the spatial dimension :returns: Dict[int, tuple] .. py:method:: _demo_item(self, dims=(4, 4), rng=None) Create an input that satisfies this spec :returns: an item like it might appear when its returned from the `__getitem__` method of a :class:`torch...Dataset`. :rtype: dict .. rubric:: Example >>> dims = (1, 1) >>> ChannelSpec.coerce(3)._demo_item(dims, rng=0) >>> ChannelSpec.coerce('r|g|b|disaprity')._demo_item(dims, rng=0) >>> ChannelSpec.coerce('rgb|disaprity')._demo_item(dims, rng=0) >>> ChannelSpec.coerce('rgb,disaprity')._demo_item(dims, rng=0) >>> ChannelSpec.coerce('rgb')._demo_item(dims, rng=0) >>> ChannelSpec.coerce('gray')._demo_item(dims, rng=0) .. py:method:: encode(self, item, axis=0, mode=1) Given a dictionary containing preloaded components of the network inputs, build a concatenated (fused) network representations of each input stream. :Parameters: * **item** (*Dict[str, Tensor]*) -- a batch item containing unfused parts. each key should be a single-stream (optionally early fused) channel key. * **axis** (*int, default=0*) -- concatenation dimension :returns: mapping between input stream and its early fused tensor input. :rtype: Dict[str, Tensor] .. rubric:: Example >>> from kwcoco.channel_spec import * # NOQA >>> import numpy as np >>> dims = (4, 4) >>> item = { >>> 'rgb': np.random.rand(3, *dims), >>> 'disparity': np.random.rand(1, *dims), >>> 'flowx': np.random.rand(1, *dims), >>> 'flowy': np.random.rand(1, *dims), >>> } >>> # Complex Case >>> self = ChannelSpec('rgb,disparity,rgb|disparity|flowx|flowy,flowx|flowy') >>> fused = self.encode(item) >>> input_shapes = ub.map_vals(lambda x: x.shape, fused) >>> print('input_shapes = {}'.format(ub.repr2(input_shapes, nl=1))) >>> # Simpler case >>> self = ChannelSpec('rgb|disparity') >>> fused = self.encode(item) >>> input_shapes = ub.map_vals(lambda x: x.shape, fused) >>> print('input_shapes = {}'.format(ub.repr2(input_shapes, nl=1))) .. rubric:: Example >>> # Case where we have to break up early fused data >>> import numpy as np >>> dims = (40, 40) >>> item = { >>> 'rgb|disparity': np.random.rand(4, *dims), >>> 'flowx': np.random.rand(1, *dims), >>> 'flowy': np.random.rand(1, *dims), >>> } >>> # Complex Case >>> self = ChannelSpec('rgb,disparity,rgb|disparity,rgb|disparity|flowx|flowy,flowx|flowy,flowx,disparity') >>> inputs = self.encode(item) >>> input_shapes = ub.map_vals(lambda x: x.shape, inputs) >>> print('input_shapes = {}'.format(ub.repr2(input_shapes, nl=1))) >>> # xdoctest: +REQUIRES(--bench) >>> #self = ChannelSpec('rgb|disparity,flowx|flowy') >>> import timerit >>> ti = timerit.Timerit(100, bestof=10, verbose=2) >>> for timer in ti.reset('mode=simple'): >>> with timer: >>> inputs = self.encode(item, mode=0) >>> for timer in ti.reset('mode=minimize-concat'): >>> with timer: >>> inputs = self.encode(item, mode=1) .. py:method:: decode(self, inputs, axis=1) break an early fused item into its components :Parameters: * **inputs** (*Dict[str, Tensor]*) -- dictionary of components * **axis** (*int, default=1*) -- channel dimension .. rubric:: Example >>> from kwcoco.channel_spec import * # NOQA >>> import numpy as np >>> dims = (4, 4) >>> item_components = { >>> 'rgb': np.random.rand(3, *dims), >>> 'ir': np.random.rand(1, *dims), >>> } >>> self = ChannelSpec('rgb|ir') >>> item_encoded = self.encode(item_components) >>> batch = {k: np.concatenate([v[None, :], v[None, :]], axis=0) ... for k, v in item_encoded.items()} >>> components = self.decode(batch) .. rubric:: Example >>> # xdoctest: +REQUIRES(module:netharn, module:torch) >>> import torch >>> import numpy as np >>> dims = (4, 4) >>> components = { >>> 'rgb': np.random.rand(3, *dims), >>> 'ir': np.random.rand(1, *dims), >>> } >>> components = ub.map_vals(torch.from_numpy, components) >>> self = ChannelSpec('rgb|ir') >>> encoded = self.encode(components) >>> from netharn.data import data_containers >>> item = {k: data_containers.ItemContainer(v, stack=True) >>> for k, v in encoded.items()} >>> batch = data_containers.container_collate([item, item]) >>> components = self.decode(batch) .. py:method:: component_indices(self, axis=2) Look up component indices within fused streams .. rubric:: Example >>> dims = (4, 4) >>> inputs = ['flowx', 'flowy', 'disparity'] >>> self = ChannelSpec('disparity,flowx|flowy') >>> component_indices = self.component_indices() >>> print('component_indices = {}'.format(ub.repr2(component_indices, nl=1))) component_indices = { 'disparity': ('disparity', (slice(None, None, None), slice(None, None, None), slice(0, 1, None))), 'flowx': ('flowx|flowy', (slice(None, None, None), slice(None, None, None), slice(0, 1, None))), 'flowy': ('flowx|flowy', (slice(None, None, None), slice(None, None, None), slice(1, 2, None))), } .. py:class:: FusedChannelSpec(parsed, _is_normalized=False) Bases: :py:obj:`BaseChannelSpec` A specific type of channel spec with only one early fused stream. The channels in this stream are non-communative Behaves like a list of atomic-channel codes (which may represent more than 1 channel), normalized codes always represent exactly 1 channel. .. note:: This class name and API is in flux and subject to change. .. todo:: A special code indicating a name and some number of bands that that names contains, this would primarilly be used for large numbers of channels produced by a network. Like: resnet_d35d060_L5:512 or resnet_d35d060_L5[:512] might refer to a very specific (hashed) set of resnet parameters with 512 bands maybe we can do something slicly like: resnet_d35d060_L5[A:B] resnet_d35d060_L5:A:B Do we want to "just store the code" and allow for parsing later? Or do we want to ensure the serialization is parsed before we construct the data structure? .. rubric:: Example >>> from kwcoco.channel_spec import * # NOQA >>> import pickle >>> self = FusedChannelSpec.coerce(3) >>> recon = pickle.loads(pickle.dumps(self)) >>> self = ChannelSpec.coerce('a|b,c|d') >>> recon = pickle.loads(pickle.dumps(self)) .. py:attribute:: _alias_lut .. py:attribute:: _memo .. py:attribute:: _size_lut .. py:method:: __len__(self) .. py:method:: __getitem__(self, index) .. py:method:: concat(cls, items) :classmethod: .. py:method:: spec(self) The string encodeing of this spec :returns: str .. py:method:: unique(self) .. py:method:: parse(cls, spec) :classmethod: .. py:method:: coerce(cls, data) :classmethod: .. rubric:: Example >>> from kwcoco.channel_spec import * # NOQA >>> FusedChannelSpec.coerce(['a', 'b', 'c']) >>> FusedChannelSpec.coerce('a|b|c') >>> FusedChannelSpec.coerce(3) >>> FusedChannelSpec.coerce(FusedChannelSpec(['a'])) >>> assert FusedChannelSpec.coerce('').numel() == 0 .. py:method:: concise(self) Shorted the channel spec by de-normaliz slice syntax :returns: concise spec :rtype: FusedChannelSpec .. rubric:: Example >>> from kwcoco.channel_spec import * # NOQA >>> self = FusedChannelSpec.coerce( >>> 'b|a|a.0|a.1|a.2|a.5|c|a.8|a.9|b.0:3|c.0') >>> short = self.concise() >>> long = short.normalize() >>> numels = [c.numel() for c in [self, short, long]] >>> print('self.spec = {!r}'.format(self.spec)) >>> print('short.spec = {!r}'.format(short.spec)) >>> print('long.spec = {!r}'.format(long.spec)) >>> print('numels = {!r}'.format(numels)) self.spec = 'b|a|a.0|a.1|a.2|a.5|c|a.8|a.9|b.0:3|c.0' short.spec = 'b|a|a:3|a.5|c|a.8:10|b:3|c.0' long.spec = 'b|a|a.0|a.1|a.2|a.5|c|a.8|a.9|b.0|b.1|b.2|c.0' numels = [13, 13, 13] >>> assert long.concise().spec == short.spec .. py:method:: normalize(self) Replace aliases with explicit single-band-per-code specs :returns: normalize spec :rtype: FusedChannelSpec .. rubric:: Example >>> from kwcoco.channel_spec import * # NOQA >>> self = FusedChannelSpec.coerce('b1|b2|b3|rgb') >>> normed = self.normalize() >>> print('self = {}'.format(self)) >>> print('normed = {}'.format(normed)) self = normed = >>> self = FusedChannelSpec.coerce('B:1:11') >>> normed = self.normalize() >>> print('self = {}'.format(self)) >>> print('normed = {}'.format(normed)) self = normed = >>> self = FusedChannelSpec.coerce('B.1:11') >>> normed = self.normalize() >>> print('self = {}'.format(self)) >>> print('normed = {}'.format(normed)) self = normed = .. py:method:: numel(self) Total number of channels in this spec .. py:method:: sizes(self) Returns a list indicating the size of each atomic code :returns: List[int] .. rubric:: Example >>> from kwcoco.channel_spec import * # NOQA >>> self = FusedChannelSpec.coerce('b1|Z:3|b2|b3|rgb') >>> self.sizes() [1, 3, 1, 1, 3] >>> assert(FusedChannelSpec.parse('a.0').numel()) == 1 >>> assert(FusedChannelSpec.parse('a:0').numel()) == 0 >>> assert(FusedChannelSpec.parse('a:1').numel()) == 1 .. py:method:: __contains__(self, key) .. rubric:: Example >>> FCS = FusedChannelSpec.coerce >>> 'disparity' in FCS('rgb|disparity|flowx|flowy') True >>> 'gray' in FCS('rgb|disparity|flowx|flowy') False .. py:method:: code_list(self) Return the expanded code list .. py:method:: as_list(self) .. py:method:: as_oset(self) .. py:method:: as_set(self) .. py:method:: as_path(self) Returns a string suitable for use in a path. Note, this may no longer be a valid channel spec .. py:method:: __set__(self) .. py:method:: difference(self, other) Set difference .. rubric:: Example >>> FCS = FusedChannelSpec.coerce >>> self = FCS('rgb|disparity|flowx|flowy') >>> other = FCS('r|b') >>> self.difference(other) >>> other = FCS('flowx') >>> self.difference(other) >>> FCS = FusedChannelSpec.coerce >>> assert len((FCS('a') - {'a'}).parsed) == 0 >>> assert len((FCS('a.0:3') - {'a.0'}).parsed) == 2 .. py:method:: intersection(self, other) .. rubric:: Example >>> FCS = FusedChannelSpec.coerce >>> self = FCS('rgb|disparity|flowx|flowy') >>> other = FCS('r|b|XX') >>> self.intersection(other) .. py:method:: union(self, other) .. rubric:: Example >>> from kwcoco.channel_spec import * # NOQA >>> FCS = FusedChannelSpec.coerce >>> self = FCS('rgb|disparity|flowx|flowy') >>> other = FCS('r|b|XX') >>> self.union(other) .. py:method:: issubset(self, other) .. py:method:: issuperset(self, other) .. py:method:: component_indices(self, axis=2) Look up component indices within this stream .. rubric:: Example >>> FCS = FusedChannelSpec.coerce >>> self = FCS('disparity|rgb|flowx|flowy') >>> component_indices = self.component_indices() >>> print('component_indices = {}'.format(ub.repr2(component_indices, nl=1))) component_indices = { 'disparity': (slice(...), slice(...), slice(0, 1, None)), 'flowx': (slice(...), slice(...), slice(4, 5, None)), 'flowy': (slice(...), slice(...), slice(5, 6, None)), 'rgb': (slice(...), slice(...), slice(1, 4, None)), } .. py:method:: streams(self) Idempotence with :func:`ChannelSpec.streams` .. py:method:: fuse(self) Idempotence with :func:`ChannelSpec.streams`