datumaro.components.dataset

Functions

eager_mode([new_mode, dataset])

Classes

Dataset([source, infos, categories, ...])	Represents a dataset, contains metainfo about labels and dataset items.
DatasetSubset(parent, name)
StreamDataset([source, infos, categories, ...])

class datumaro.components.dataset.Dataset(source: IDataset | None = None, *, infos: Dict[str, Any] | None = None, categories: Dict[AnnotationType, Categories] | None = None, media_type: Type[MediaElement] | None = None, ann_types: Set[AnnotationType] | None = None, env: Environment | None = None)

Bases: IDataset

Represents a dataset, contains metainfo about labels and dataset items. Provides iteration and access options to dataset elements.

By default, all operations are done lazily, it can be changed by modifying the eager property and by using the eager_mode context manager.

Dataset is supposed to have a single media type for its items. If the dataset is filled manually or from extractors, and media type does not match, an error is raised.

classmethod from_iterable(iterable: ~typing.Iterable[~datumaro.components.dataset_base.DatasetItem], infos: ~typing.Dict[str, ~typing.Any] | None = None, categories: ~typing.Dict[~datumaro.components.annotation.AnnotationType, ~datumaro.components.annotation.Categories] | ~typing.List[str] | None = None, *, env: ~datumaro.components.environment.Environment | None = None, media_type: ~typing.Type[~datumaro.components.media.MediaElement] = <class 'datumaro.components.media.Image'>, ann_types: ~typing.Set[~datumaro.components.annotation.AnnotationType] | None = []) → Dataset

Creates a new dataset from an iterable object producing dataset items - a generator, a list etc. It is a convenient way to create and fill a custom dataset.

Parameters:

• iterable – An iterable which returns dataset items

• infos – A dictionary of the dataset specific information

• categories – A simple list of labels or complete information about labels. If not specified, an empty list of labels is assumed.

• media_type – Media type for the dataset items. If the sequence contains items with mismatching media type, an error is raised during caching

• env – A context for plugins, which will be used for this dataset. If not specified, the builtin plugins will be used.

Returns:

A new dataset with specified contents

Return type:

dataset

classmethod from_extractors(*sources: IDataset, env: Environment | None = None, merge_policy: str = 'exact') → Dataset

Creates a new dataset from one or several `Extractor`s.

In case of a single input, creates a lazy wrapper around the input. In case of several inputs, merges them and caches the resulting dataset.

Parameters:

• sources – one or many input extractors

• env – A context for plugins, which will be used for this dataset. If not specified, the builtin plugins will be used.

• merge_policy – Policy on how to merge multiple datasets. Possible options are “exact”, “intersect”, and “union”.

Returns:

A new dataset with contents produced by input extractors

Return type:

dataset

define_infos(infos: Dict[str, Any]) → None

define_categories(categories: Dict[AnnotationType, Categories]) → None

init_cache() → None

get_subset(name) → DatasetSubset

subsets() → Dict[str, DatasetSubset]

Enumerates subsets in the dataset. Each subset can be a dataset itself.

infos() → Dict[str, Any]

Returns meta-info of dataset.

categories() → Dict[AnnotationType, Categories]

Returns metainfo about dataset labels.

media_type() → Type[MediaElement]

Returns media type of the dataset items.

All the items are supposed to have the same media type. Supposed to be constant and known immediately after the object construction (i.e. doesn’t require dataset iteration).

ann_types() → Set[AnnotationType]

Returns available task type from dataset annotation types.

get(id: str, subset: str | None = None) → DatasetItem | None

Provides random access to dataset items.

get_annotated_items()

get_annotations()

get_datasetitem_by_path(path)

get_label_cat_names()

get_subset_info() → str

get_infos() → Tuple[str]

get_categories_info() → Tuple[str]

put(item: DatasetItem, id: str | None = None, subset: str | None = None) → None

remove(id: str, subset: str | None = None) → None

filter(expr: str, *, filter_annotations: bool = False, remove_empty: bool = False) → Dataset

filter(filter_func: Callable[[DatasetItem], bool] | Callable[[DatasetItem, Annotation], bool], *, filter_annotations: bool = False, remove_empty: bool = False) → Dataset

update(source: DatasetPatch | IDataset | Iterable[DatasetItem]) → Dataset

Updates items of the current dataset from another dataset or an iterable (the source). Items from the source overwrite matching items in the current dataset. Unmatched items are just appended.

If the source is a DatasetPatch, the removed items in the patch will be removed in the current dataset.

If the source is a dataset, labels are matched. If the labels match, but the order is different, the annotation labels will be remapped to the current dataset label order during updating.

Returns: self

transform(method: str | Type[Transform], **kwargs) → Dataset

Applies some function to dataset items.

Results are stored in-place. Modifications are applied lazily. Transforms are not allowed to change media type of dataset items.

Parameters:

• method – The transformation to be applied to the dataset. If a string is passed, it is treated as a plugin name, which is searched for in the dataset environment.

• **kwargs – Parameters for the transformation

Returns: self

run_model(model: Launcher | Type[ModelTransform], *, batch_size: int = 1, append_annotation: bool = False, num_workers: int = 0, **kwargs) → Dataset

Applies a model to dataset items’ media and produces a dataset with media and annotations.

Parameters:

• model – The model to be applied to the dataset

• batch_size – The number of dataset items processed simultaneously by the model

• append_annotation – Whether append new annotation to existed annotations

• num_workers – The number of worker threads to use for parallel inference. Set to 0 for single-process mode. Default is 0.

• **kwargs – Parameters for the model

Returns: self

select(pred: Callable[[DatasetItem], bool]) → Dataset

property data_path: str | None

property format: str | None

property options: Dict[str, Any]

property is_modified: bool

get_patch() → DatasetPatch

property env: Environment

property is_cache_initialized: bool

property is_eager: bool

property is_bound: bool

bind(path: str, format: str | None = None, *, options: Dict[str, Any] | None = None) → None

Binds the dataset to a speific directory. Allows to set default saving parameters.

The following saves will be done to this directory by default and will use the saved parameters.

flush_changes()

export(save_dir: str, format: str | Type[Exporter], *, progress_reporter: ProgressReporter | None = None, error_policy: ExportErrorPolicy | None = None, **kwargs) → None

Saves the dataset in some format.

Parameters:

• save_dir – The output directory

• format – The desired output format. If a string is passed, it is treated as a plugin name, which is searched for in the dataset environment.

• progress_reporter – An object to report progress

• error_policy – An object to report format-related errors

• **kwargs – Parameters for the format

save(save_dir: str | None = None, **kwargs) → None

classmethod load(path: str, **kwargs) → Dataset

classmethod import_from(path: str, format: str | None = None, *, env: Environment | None = None, progress_reporter: ProgressReporter | None = None, error_policy: ImportErrorPolicy | None = None, **kwargs) → Dataset

Creates a Dataset instance from a dataset on the disk.

Parameters:

• path (path - The input file or directory) –

• format. (format - Dataset) – If a string is passed, it is treated as a plugin name, which is searched for in the env plugin context. If not set, will try to detect automatically, using the env plugin context.

• set (env - A plugin collection. If not) –

• used (the built-in plugins are) –

• progress. (progress_reporter - An object to report) – Implies earger loading.

• errors. (error_policy - An object to report format-related) – Implies earger loading.

• format (**kwargs - Parameters for the) –

static detect(path: str, *, env: Environment | None = None, depth: int = 2) → str

Attempts to detect dataset format of a given directory.

This function tries to detect a single format and fails if it’s not possible. Check Environment.detect_dataset() for a function that reports status for each format checked.

Parameters:

• path – The directory to check

• depth – The maximum depth for recursive search

• env – A plugin collection. If not set, the built-in plugins are used

property is_stream: bool

Boolean indicating whether the dataset is a stream

If the dataset is a stream, the dataset item is generated on demand from its iterator.

clone() → Dataset

Create a deep copy of this dataset.

Returns:

A cloned instance of the Dataset.

datumaro.components.dataset.eager_mode(new_mode: bool = True, dataset: Dataset | None = None) → None

class datumaro.components.dataset.Annotation(*, id: int = 0, attributes: Dict[str, Any] = _Nothing.NOTHING, group: int = 0, object_id: int = -1)

Bases: object

A base annotation class.

Derived classes must define the ‘_type’ class variable with a value from the AnnotationType enum.

Method generated by attrs for class Annotation.

id: int

attributes: Dict[str, Any]

group: int

object_id: int

property type: AnnotationType

as_dict() → Dict[str, Any]

Returns a dictionary { field_name: value }

wrap(**kwargs)

Returns a modified copy of the object

class datumaro.components.dataset.AnnotationType(value)

Bases: IntEnum

An enumeration.

unknown = 0

label = 1

mask = 2

points = 3

polygon = 4

polyline = 5

bbox = 6

caption = 7

cuboid_3d = 8

super_resolution_annotation = 9

depth_annotation = 10

ellipse = 11

hash_key = 12

feature_vector = 13

tabular = 14

rotated_bbox = 15

cuboid_2d = 16

class datumaro.components.dataset.DatasetBase(*, length: int | None = None, subsets: ~typing.Sequence[str] | None = None, media_type: ~typing.Type[~datumaro.components.media.MediaElement] = <class 'datumaro.components.media.Image'>, ann_types: ~typing.List[~datumaro.components.annotation.AnnotationType] | None = None, ctx: ~datumaro.components.contexts.importer.ImportContext | None = None)

Bases: _DatasetBase, CliPlugin

A base class for user-defined and built-in extractors. Should be used in cases, where SubsetBase is not enough, or its use makes problems with performance, implementation etc.

media_type()

Returns media type of the dataset items.

All the items are supposed to have the same media type. Supposed to be constant and known immediately after the object construction (i.e. doesn’t require dataset iteration).

ann_types()

Returns available task type from dataset annotation types.

exception datumaro.components.dataset.DatasetImportError

Bases: DatumaroError

class datumaro.components.dataset.DatasetItem(id: str, *, subset: str | None = None, media: str | MediaElement | None = None, annotations: List[Annotation] | None = None, attributes: Dict[str, Any] | None = None)

Bases: object

id: str

subset: str

media: MediaElement | None

annotations: Annotations

attributes: Dict[str, Any]

wrap(**kwargs)

media_as(t: Type[T]) → T

class datumaro.components.dataset.DatasetItemStorageDatasetView(parent: DatasetItemStorage, infos: Dict[str, Any], categories: Dict[AnnotationType, Categories], media_type: Type[MediaElement] | None, ann_types: Set[AnnotationType] | None)

Bases: IDataset

class Subset(parent: DatasetItemStorageDatasetView, name: str)

Bases: IDataset

put(item)

get(id, subset=None)

Provides random access to dataset items.

remove(id, subset=None)

get_subset(name)

subsets()

Enumerates subsets in the dataset. Each subset can be a dataset itself.

infos()

Returns meta-info of dataset.

categories()

Returns metainfo about dataset labels.

media_type()

Returns media type of the dataset items.

All the items are supposed to have the same media type. Supposed to be constant and known immediately after the object construction (i.e. doesn’t require dataset iteration).

ann_types()

Returns available task type from dataset annotation types.

infos()

Returns meta-info of dataset.

categories()

Returns metainfo about dataset labels.

get_subset(name)

subsets()

Enumerates subsets in the dataset. Each subset can be a dataset itself.

get(id, subset=None)

Provides random access to dataset items.

media_type()

Returns media type of the dataset items.

All the items are supposed to have the same media type. Supposed to be constant and known immediately after the object construction (i.e. doesn’t require dataset iteration).

ann_types()

Returns available task type from dataset annotation types.

class datumaro.components.dataset.DatasetPatch(data: DatasetItemStorage, infos: Dict[str, Any], categories: Dict[AnnotationType, Categories], updated_items: Dict[Tuple[str, str], ItemStatus], updated_subsets: Dict[str, ItemStatus] | None = None)

Bases: object

class DatasetPatchWrapper(patch: DatasetPatch, parent: IDataset)

Bases: DatasetItemStorageDatasetView

subsets()

Enumerates subsets in the dataset. Each subset can be a dataset itself.

property updated_subsets: Dict[str, ItemStatus]

as_dataset(parent: IDataset) → IDataset

class datumaro.components.dataset.DatasetStorage(source: IDataset | DatasetItemStorage, infos: Dict[str, Any] | None = None, categories: Dict[AnnotationType, Categories] | None = None, media_type: Type[MediaElement] | None = None, ann_types: Set[AnnotationType] | None = None)

Bases: IDataset

is_cache_initialized() → bool

init_cache() → None

infos() → Dict[str, Any]

Returns meta-info of dataset.

define_infos(infos: Dict[str, Any])

categories() → Dict[AnnotationType, Categories]

Returns metainfo about dataset labels.

define_categories(categories: Dict[AnnotationType, Categories])

media_type() → Type[MediaElement]

Returns media type of the dataset items.

All the items are supposed to have the same media type. Supposed to be constant and known immediately after the object construction (i.e. doesn’t require dataset iteration).

ann_types() → Set[AnnotationType]

Returns available task type from dataset annotation types.

put(item: DatasetItem) → None

get(id: str, subset: str | None = None) → DatasetItem | None

Provides random access to dataset items.

remove(id: str, subset: str | None = None) → None

get_subset(name: str) → IDataset

subsets() → Dict[str, IDataset]

Enumerates subsets in the dataset. Each subset can be a dataset itself.

get_annotated_items() → int

get_annotations() → int

get_datasetitem_by_path(path: str) → DatasetItem | None

transform(method: Type[Transform], *args, **kwargs) → None

has_updated_items()

get_patch() → DatasetPatch

flush_changes()

update(source: DatasetPatch | IDataset | Iterable[DatasetItem])

class datumaro.components.dataset.DatasetSubset(parent: Dataset, name: str)

Bases: IDataset

put(item)

get(id, subset=None)

Provides random access to dataset items.

remove(id, subset=None)

get_subset(name)

subsets()

Enumerates subsets in the dataset. Each subset can be a dataset itself.

infos()

Returns meta-info of dataset.

categories()

Returns metainfo about dataset labels.

media_type()

Returns media type of the dataset items.

All the items are supposed to have the same media type. Supposed to be constant and known immediately after the object construction (i.e. doesn’t require dataset iteration).

ann_types()

Returns available task type from dataset annotation types.

get_annotated_items()

get_annotations()

as_dataset() → Dataset

exception datumaro.components.dataset.DatumaroError

Bases: Exception

class datumaro.components.dataset.Environment(use_lazy_import: bool = True)

Bases: object

property extractors: DatasetBaseRegistry

property importers: ImporterRegistry

property launchers: LauncherRegistry

property exporters: ExporterRegistry

property generators: GeneratorRegistry

property transforms: TransformRegistry

property validators: ValidatorRegistry

load_plugins(plugins_dir)

register_plugins(plugins)

make_extractor(name, *args, **kwargs)

make_importer(name, *args, **kwargs)

make_launcher(name, *args, **kwargs)

make_exporter(name, *args, **kwargs)

make_transform(name, *args, **kwargs)

is_format_known(name)

detect_dataset(path: str, depth: int = 1, rejection_callback: Callable[[str, RejectionReason, str], None] | None = None) → List[str]

classmethod merge(envs: Sequence[Environment]) → Environment

classmethod release_builtin_plugins()

class datumaro.components.dataset.ExportContext(progress_reporter=None, error_policy=None)

Bases: object

Method generated by attrs for class ExportContext.

progress_reporter: ProgressReporter

error_policy: ExportErrorPolicy

class datumaro.components.dataset.ExportErrorPolicy

Bases: object

report_item_error(error: Exception, *, item_id: Tuple[str, str]) → None

Allows to report a problem with a dataset item. If this function returns, the converter must skip the item.

report_annotation_error(error: Exception, *, item_id: Tuple[str, str]) → None

Allows to report a problem with a dataset item annotation. If this function returns, the converter must skip the annotation.

fail(error: Exception) → NoReturn

class datumaro.components.dataset.Exporter(extractor: IDataset, save_dir: str, *, save_media: bool | None = None, image_ext: str | None = None, default_image_ext: str | None = None, save_dataset_meta: bool = False, save_hashkey_meta: bool = False, stream: bool = False, ctx: ExportContext | None = None)

Bases: CliPlugin

DEFAULT_IMAGE_EXT = None

classmethod build_cmdline_parser(**kwargs)

classmethod convert(extractor, save_dir, **options)

classmethod patch(dataset, patch, save_dir, **options)

apply()

Execute the data-format conversion

property can_stream: bool

Flag to indicate whether the exporter can export the dataset in a stream manner or not.

class datumaro.components.dataset.IDataset

Bases: object

subsets() → Dict[str, IDataset]

Enumerates subsets in the dataset. Each subset can be a dataset itself.

get_subset(name) → IDataset

infos() → Dict[str, Any]

Returns meta-info of dataset.

categories() → Dict[AnnotationType, Categories]

Returns metainfo about dataset labels.

get(id: str, subset: str | None = None) → DatasetItem | None

Provides random access to dataset items.

media_type() → Type[MediaElement]

Returns media type of the dataset items.

All the items are supposed to have the same media type. Supposed to be constant and known immediately after the object construction (i.e. doesn’t require dataset iteration).

ann_types() → List[AnnotationType]

Returns available task type from dataset annotation types.

property is_stream: bool

Boolean indicating whether the dataset is a stream

If the dataset is a stream, the dataset item is generated on demand from its iterator.

class datumaro.components.dataset.Image(size: Tuple[int, int] | None = None, ext: str | None = None, *args, **kwargs)

Bases: MediaElement[ndarray]

classmethod from_file(path: str, *args, **kwargs)

classmethod from_numpy(data: ndarray | Callable[[], ndarray], *args, **kwargs)

classmethod from_bytes(data: bytes | Callable[[], bytes], *args, **kwargs)

property has_size: bool

Indicates that size info is cached and won’t require image loading

property size: Tuple[int, int] | None

Returns (H, W)

property ext: str | None

Media file extension (with the leading dot)

set_crypter(crypter: Crypter)

class datumaro.components.dataset.ImportContext(progress_reporter=None, error_policy=None)

Bases: object

Method generated by attrs for class ImportContext.

progress_reporter: ProgressReporter

error_policy: ImportErrorPolicy

class datumaro.components.dataset.ImportErrorPolicy

Bases: object

report_item_error(error: Exception, *, item_id: Tuple[str, str]) → None

Allows to report a problem with a dataset item. If this function returns, the extractor must skip the item.

report_annotation_error(error: Exception, *, item_id: Tuple[str, str]) → None

Allows to report a problem with a dataset item annotation. If this function returns, the extractor must skip the annotation.

fail(error: Exception) → NoReturn

class datumaro.components.dataset.ItemTransform(extractor: IDataset)

Bases: Transform

transform_item(item: DatasetItem) → DatasetItem | None

Returns a modified copy of the input item.

Avoid changing and returning the input item, because it can lead to unexpected problems. Use wrap_item() or item.wrap() to simplify copying.

class datumaro.components.dataset.LabelCategories(items: List[str] = _Nothing.NOTHING, label_groups: List[LabelGroup] = _Nothing.NOTHING, *, attributes: Set[str] = _Nothing.NOTHING)

Bases: Categories

Method generated by attrs for class LabelCategories.

class Category(name, parent: str = '', attributes: Set[str] = _Nothing.NOTHING)

Bases: object

Method generated by attrs for class LabelCategories.Category.

name: str

parent: str

attributes: Set[str]

class LabelGroup(name, labels: List[str] = [], group_type: GroupType = GroupType.EXCLUSIVE)

Bases: object

Method generated by attrs for class LabelCategories.LabelGroup.

name: str

labels: List[str]

group_type: GroupType

items: List[str]

label_groups: List[LabelGroup]

classmethod from_iterable(iterable: Iterable[str | Tuple[str] | Tuple[str, str] | Tuple[str, str, List[str]]]) → LabelCategories

Creates a LabelCategories from iterable.

Parameters:

iterable –

This iterable object can be:

• a list of str - will be interpreted as list of Category names

• a list of positional arguments - will generate Categories with these arguments

Returns: a LabelCategories object

add(name: str, parent: str | None = None, attributes: Set[str] | None = None) → int

add_label_group(name: str, labels: List[str], group_type: GroupType) → int

find(name: str) → Tuple[int | None, Category | None]

class datumaro.components.dataset.Launcher(model_dir: str | None = None)

Bases: CliPlugin

preprocess(item: DatasetItem) → Tuple[ndarray | Dict[str, ndarray], PrepInfo]

Preprocess single dataset item before launch()

There are two output types:

1. The output is np.ndarray. For example, it can be image data as np.ndarray with BGR format (H, W, C). In this step, you usually implement resizing, normalizing, or color channel conversion for your launcher (or model).

2. The output is Dict[str, np.ndarray]. For example, it can be image and text pairs. Therefore, this can be used for the model having multi modality for image and text inputs.

infer(inputs: Dict[str, ndarray]) → List[Dict[str, ndarray] | List[Dict[str, ndarray]]]

infer(inputs: ndarray) → List[Dict[str, ndarray] | List[Dict[str, ndarray]]]

postprocess(pred: Dict[str, ndarray] | List[Dict[str, ndarray]], info: PrepInfo) → List[Annotation]

launch(batch: Sequence[DatasetItem], stack: bool = True) → List[List[Annotation]]

Launch to obtain the inference outputs of items.

Parameters:

• inputs – batch of Datasetitems

• stack – If true, launch inference for the stacked input for the batch-wise dimension Otherwise, launch inference for each input.

Returns:

A list of annotation list. Each annotation list is mapped to the input DatasetItem. These annotation list are pseudo-labels obtained by the model inference.

infos()

categories()

type_check(item: DatasetItem) → bool

Check the media type of dataset item.

If False, the item is excluded from the input batch.

class datumaro.components.dataset.MediaElement(crypter: ~datumaro.components.crypter.Crypter = <datumaro.components.crypter.NullCrypter object>, *args, **kwargs)

Bases: Generic[AnyData]

as_dict() → Dict[str, Any]

from_self(**kwargs)

property is_encrypted: bool

set_crypter(crypter: Crypter)

property type: MediaType

property data: AnyData | None

property has_data: bool

property bytes: bytes | None

save(fp: str | ~io.IOBase, crypter: ~datumaro.components.crypter.Crypter = <datumaro.components.crypter.NullCrypter object>)

class datumaro.components.dataset.ModelTransform(extractor: IDataset, launcher: Launcher, batch_size: int = 1, append_annotation: bool = False, num_workers: int = 0)

Bases: Transform

A transformation class for applying a model’s inference to dataset items.

This class takes an dataset, a launcher, and other optional parameters to transform the dataset item from the model outputs by the launcher. It can process items using multiple processes if specified, making it suitable for parallelized inference tasks.

Parameters:

• extractor – The dataset extractor to obtain items from.

• launcher – The launcher responsible for model inference.

• batch_size – The batch size for processing items. Default is 1.

• append_annotation – Whether to append inference annotations to existing annotations. Default is False.

• num_workers – The number of worker threads to use for parallel inference. Set to 0 for single-process mode. Default is 0.

get_subset(name)

infos()

Returns meta-info of dataset.

categories()

Returns metainfo about dataset labels.

transform_item(item)

exception datumaro.components.dataset.MultipleFormatsMatchError(formats)

Bases: DatasetImportError

Method generated by attrs for class MultipleFormatsMatchError.

formats

exception datumaro.components.dataset.NoMatchingFormatsError

Bases: DatasetImportError

class datumaro.components.dataset.NullProgressReporter

Bases: ProgressReporter

property period: float

Returns reporting period.

For example, 0.1 would mean every 10%.

property interval: float

Returns reporting time interval in second.

start(total: int, *, desc: str | None = None)

Initializes the progress bar

report_status(progress: int)

Updates the progress bar

iter(iterable: Iterable[T], *, total: int | None = None, desc: str | None = None) → Iterable[T]

Traverses the iterable and reports progress simultaneously.

Starts and finishes the progress bar automatically.

Parameters:

• iterable – An iterable to be traversed

• total – The expected number of iterations. If not provided, will try to use iterable.__len__.

• desc – The status message

Returns:

An iterable over elements of the input sequence

split(count: int) → Tuple[ProgressReporter]

Splits the progress bar into few independent parts. In case of 0 must return an empty tuple.

This class is supposed to manage the state of children progress bars and release of their resources, if necessary.

class datumaro.components.dataset.ProgressReporter

Bases: object

Only one set of methods must be called:

• start - report_status - finish

• iter

• split

This class is supposed to manage the state of children progress bars and release of their resources, if necessary.

property period: float

Returns reporting period.

For example, 0.1 would mean every 10%.

property interval: float

Returns reporting time interval in second.

start(total: int, *, desc: str | None = None)

Initializes the progress bar

report_status(progress: int)

Updates the progress bar

finish()

Finishes the progress bar

iter(iterable: Iterable[T], *, total: int | None = None, desc: str | None = None) → Iterable[T]

Traverses the iterable and reports progress simultaneously.

Starts and finishes the progress bar automatically.

Parameters:

• iterable – An iterable to be traversed

• total – The expected number of iterations. If not provided, will try to use iterable.__len__.

• desc – The status message

Returns:

An iterable over elements of the input sequence

split(count: int) → Tuple[ProgressReporter, ...]

Splits the progress bar into few independent parts. In case of 0 must return an empty tuple.

This class is supposed to manage the state of children progress bars and release of their resources, if necessary.

class datumaro.components.dataset.Source(config=None)

Bases: Config

property is_generated: bool

class datumaro.components.dataset.StreamDataset(source: IDataset | None = None, *, infos: Dict[str, Any] | None = None, categories: Dict[AnnotationType, Categories] | None = None, media_type: Type[MediaElement] | None = None, ann_types: Set[AnnotationType] | None = None, env: Environment | None = None)

Bases: Dataset

property is_eager: bool

classmethod from_extractors(*sources: IDataset, env: Environment | None = None, merge_policy: str = 'exact') → Dataset

Creates a new dataset from one or several `Extractor`s.

In case of a single input, creates a lazy wrapper around the input. In case of several inputs, unifies them and caches the resulting dataset. We cannot apply regular dataset merge, since items list cannot be accessed.

Parameters:

• sources – one or many input extractors

• env – A context for plugins, which will be used for this dataset. If not specified, the builtin plugins will be used.

• merge_policy – Policy on how to merge multiple datasets. Possible options are “exact”, “intersect”, and “union”.

Returns:

A new dataset with contents produced by input extractors

Return type:

dataset

class datumaro.components.dataset.StreamDatasetStorage(source: IDataset, infos: Dict[str, Any] | None = None, categories: Dict[AnnotationType, Categories] | None = None, media_type: Type[MediaElement] | None = None, ann_types: Set[AnnotationType] | None = None)

Bases: DatasetStorage

is_cache_initialized() → bool

init_cache() → None

property stacked_transform: IDataset

put(item: DatasetItem) → None

get(id: str, subset: str | None = None) → DatasetItem | None

Provides random access to dataset items.

remove(id: str, subset: str | None = None) → None

get_subset(name: str) → IDataset

property subset_names

subsets() → Dict[str, IDataset]

Enumerates subsets in the dataset. Each subset can be a dataset itself.

transform(method: Type[Transform], *args, **kwargs) → None

get_annotated_items() → int

get_annotations() → int

get_datasetitem_by_path(path: str) → DatasetItem | None

get_patch()

flush_changes()

update(source: DatasetPatch | IDataset | Iterable[DatasetItem])

infos() → Dict[str, Any]

Returns meta-info of dataset.

categories() → Dict[AnnotationType, Categories]

Returns metainfo about dataset labels.

property is_stream: bool

Boolean indicating whether the dataset is a stream

If the dataset is a stream, the dataset item is generated on demand from its iterator.

exception datumaro.components.dataset.StreamedItemError

Bases: DatasetError

Method generated by attrs for class StreamedItemError.

class datumaro.components.dataset.TabularCategories(items: List[Category] = _Nothing.NOTHING, *, attributes: Set[str] = _Nothing.NOTHING)

Bases: Categories

Describes tabular data metainfo such as column names and types.

Method generated by attrs for class TabularCategories.

class Category(name, dtype: Type[TableDtype], labels: Set[str | int] = _Nothing.NOTHING)

Bases: object

Method generated by attrs for class TabularCategories.Category.

name: str

dtype: Type[TableDtype]

labels: Set[str | int]

items: List[Category]

classmethod from_iterable(iterable: Iterable[Tuple[str, Type[TableDtype]] | Tuple[str, Type[TableDtype], Set[str]]]) → TabularCategories

Creates a TabularCategories from iterable.

Parameters:

iterable – a list of (Category name, type) or (Category name, type, set of labels)

Returns: a TabularCategories object

add(name: str, dtype: Type[TableDtype], labels: Set[str] | None = None) → int

Add a Tabular Category.

Parameters:

• name (str) – Column name

• dtype (type) – Type of the corresponding column. (str, int, or float)

• labels (optional, set(str)) – Label values where the column can have.

Returns:

A index of added category.

Return type:

int

find(name: str) → Tuple[int | None, Category | None]

Find Category information for the given column name.

Parameters:

name (str) – Column name

Returns:

A index and Category information.

Return type:

tuple(int, Category)

class datumaro.components.dataset.Transform(extractor: IDataset)

Bases: DatasetBase, CliPlugin

A base class for dataset transformations that change dataset items or their annotations.

static wrap_item(item, **kwargs)

categories()

Returns metainfo about dataset labels.

subsets()

Enumerates subsets in the dataset. Each subset can be a dataset itself.

media_type()

Returns media type of the dataset items.

All the items are supposed to have the same media type. Supposed to be constant and known immediately after the object construction (i.e. doesn’t require dataset iteration).

infos()

Returns meta-info of dataset.

exception datumaro.components.dataset.UnknownFormatError(format)

Bases: DatumaroError

Method generated by attrs for class UnknownFormatError.

format

class datumaro.components.dataset.UserFunctionAnnotationsFilter(extractor: IDataset, filter_func: Callable[[DatasetItem, Annotation], bool], remove_empty: bool = False)

Bases: ItemTransform

Filter annotations using a user-provided Python function.

Parameters:

• extractor – Datumaro Dataset to filter.

• filter_func – A Python callable that takes DatasetItem and Annotation as its inputs and returns a boolean. If the return value is True, the Annotation will be retained. Otherwise, it is removed.

• remove_empty – If True, DatasetItem without any annotations is removed after filtering its annotations. Otherwise, do not filter DatasetItem.

Example

This is an example of removing bounding boxes sized greater than 50% of the image size:

from datumaro.components.media import Image from datumaro.components.annotation import Annotation, Bbox

def filter_func(item: DatasetItem, ann: Annotation) -> bool:

# If the annotation is not a Bbox, do not filter if not isinstance(ann, Bbox):

return False

h, w = item.media_as(Image).size image_size = h * w bbox_size = ann.h * ann.w

# Accept Bboxes smaller than 50% of the image size return bbox_size < 0.5 * image_size

filtered = UserFunctionAnnotationsFilter(

extractor=dataset, filter_func=filter_func)

# No bounding boxes with a size greater than 50% of their image filtered_items = [item for item in filtered]

transform_item(item: DatasetItem) → DatasetItem | None

Returns a modified copy of the input item.

Avoid changing and returning the input item, because it can lead to unexpected problems. Use wrap_item() or item.wrap() to simplify copying.

class datumaro.components.dataset.UserFunctionDatasetFilter(extractor: IDataset, filter_func: Callable[[DatasetItem], bool])

Bases: ItemTransform

Filter dataset items using a user-provided Python function.

Parameters:

• extractor – Datumaro Dataset to filter.

• filter_func – A Python callable that takes a DatasetItem as its input and returns a boolean. If the return value is True, that DatasetItem will be retained. Otherwise, it is removed.

Example

This is an example of filtering dataset items with images larger than 1024 pixels:

from datumaro.components.media import Image

def filter_func(item: DatasetItem) -> bool:

h, w = item.media_as(Image).size return h > 1024 or w > 1024

filtered = UserFunctionDatasetFilter(

extractor=dataset, filter_func=filter_func)

# No items with an image height or width greater than 1024 filtered_items = [item for item in filtered]

transform_item(item: DatasetItem) → DatasetItem | None

Returns a modified copy of the input item.

Avoid changing and returning the input item, because it can lead to unexpected problems. Use wrap_item() or item.wrap() to simplify copying.

class datumaro.components.dataset.XPathAnnotationsFilter(extractor: IDataset, xpath: str, remove_empty: bool = False)

Bases: ItemTransform

transform_item(item: DatasetItem) → DatasetItem | None

Returns a modified copy of the input item.

Avoid changing and returning the input item, because it can lead to unexpected problems. Use wrap_item() or item.wrap() to simplify copying.

class datumaro.components.dataset.XPathDatasetFilter(extractor: IDataset, xpath: str)

Bases: ItemTransform

transform_item(item: DatasetItem) → DatasetItem | None

Returns a modified copy of the input item.

Avoid changing and returning the input item, because it can lead to unexpected problems. Use wrap_item() or item.wrap() to simplify copying.

datumaro.components.dataset.contextmanager(func)

@contextmanager decorator.

Typical usage:

@contextmanager def some_generator(<arguments>):

<setup> try:

yield <value>

finally:

<cleanup>

This makes this:

with some_generator(<arguments>) as <variable>:

<body>

equivalent to this:

<setup> try:

<variable> = <value> <body>

finally:

<cleanup>

datumaro.components.dataset.copy(x)

Shallow copy operation on arbitrary Python objects.

See the module’s __doc__ string for more info.

datumaro.components.dataset.deepcopy(x, memo=None, _nil=[])

Deep copy operation on arbitrary Python objects.

See the module’s __doc__ string for more info.

datumaro.components.dataset.load_hash_key(path, dataset)

datumaro.components.dataset.logging_disabled(max_level=50)

datumaro.components.dataset.on_error_do(callback, *args, ignore_errors=False, kwargs=None)

datumaro.components.dataset.overload(func)

Decorator for overloaded functions/methods.

In a stub file, place two or more stub definitions for the same function in a row, each decorated with @overload. For example:

@overload def utf8(value: None) -> None: … @overload def utf8(value: bytes) -> bytes: … @overload def utf8(value: str) -> bytes: …

In a non-stub file (i.e. a regular .py file), do the same but follow it with an implementation. The implementation should not be decorated with @overload. For example:

@overload def utf8(value: None) -> None: … @overload def utf8(value: bytes) -> bytes: … @overload def utf8(value: str) -> bytes: … def utf8(value):

# implementation goes here

datumaro.components.dataset.rmtree(path: str | PathLike[str]) → None

Remove the given directory tree recursively.

Note:

We use shutil.rmtree() but adjust its behaviour to see whether files that couldn’t be deleted are read-only. Windows will not remove them in that case.

datumaro.components.dataset.scoped(func, arg_name=None)

A function decorator, which allows to do actions with the current scope, such as registering error and exit callbacks and context managers.